taosdata/TDengine | Build 3796 | source/libs/executor/src/streamfilloperator.c | Coveralls

43

TSKEY getNextWindowTs(TSKEY ts, SInterval* pInterval) {

×

44

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

×

45

  getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);

×

46

  return win.skey;

×

49

TSKEY getPrevWindowTs(TSKEY ts, SInterval* pInterval) {

×

50

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

×

51

  getNextTimeWindow(pInterval, &win, TSDB_ORDER_DESC);

×

52

  return win.skey;

×

55

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

×

56

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

×

59

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

×

60

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

×

61

    return NULL;

×

63

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

×

64

  SResultCellData* pCell = pRaw->pRowVal;

×

65

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

×

66

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

×

67

    pCell = (SResultCellData*)pData;

×

69

  return pCell;

×

72

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

×

73

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

×

74

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

×

75

    taosVariantDestroy(&pFillCol[i].fillVal);

×

77

  if (start < end) {

×

78

    taosMemoryFreeClear(pFillCol[start].pExpr);

×

80

  taosMemoryFree(pFillCol);

×

81

  return NULL;

×

84

void destroyStreamFillSupporter(SStreamFillSupporter* pFillSup) {

×

85

  if (pFillSup == NULL) {

×

86

    return;

×

88

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

×

89

  tSimpleHashCleanup(pFillSup->pResMap);

×

90

  pFillSup->pResMap = NULL;

×

91

  cleanupExprSupp(&pFillSup->notFillExprSup);

×

92

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

×

93

    taosMemoryFree(pFillSup->cur.pRowVal);

×

95

  taosMemoryFree(pFillSup->prev.pRowVal);

×

96

  taosMemoryFree(pFillSup->next.pRowVal);

×

97

  taosMemoryFree(pFillSup->nextNext.pRowVal);

×

99

  taosMemoryFree(pFillSup->pOffsetInfo);

×

100

  taosArrayDestroy(pFillSup->pResultRange);

×

101

  pFillSup->pResultRange = NULL;

×

103

  taosMemoryFree(pFillSup);

×

106

void destroySPoint(void* ptr) {

×

107

  SPoint* point = (SPoint*)ptr;

×

108

  taosMemoryFreeClear(point->val);

×

109

×

111

void destroyStreamFillLinearInfo(SStreamFillLinearInfo* pFillLinear) {

×

112

  taosArrayDestroyEx(pFillLinear->pEndPoints, destroySPoint);

×

113

  taosArrayDestroyEx(pFillLinear->pNextEndPoints, destroySPoint);

×

114

  taosMemoryFree(pFillLinear);

×

115

×

117

void destroyStreamFillInfo(SStreamFillInfo* pFillInfo) {

×

118

  if (pFillInfo == NULL) {

×

119

    return;

×

121

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

×

122

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

×

123

    taosMemoryFreeClear(pFillInfo->pResRow->pRowVal);

×

124

    taosMemoryFreeClear(pFillInfo->pResRow);

×

125

    taosMemoryFreeClear(pFillInfo->pNonFillRow->pRowVal);

×

126

    taosMemoryFreeClear(pFillInfo->pNonFillRow);

×

128

  destroyStreamFillLinearInfo(pFillInfo->pLinearInfo);

×

129

  pFillInfo->pLinearInfo = NULL;

×

131

  taosArrayDestroy(pFillInfo->delRanges);

×

132

  taosMemoryFreeClear(pFillInfo->pTempBuff);

×

133

  taosMemoryFree(pFillInfo);

×

136

void clearGroupResArray(SGroupResInfo* pGroupResInfo) {

×

137

  pGroupResInfo->freeItem = false;

×

138

  taosArrayDestroy(pGroupResInfo->pRows);

×

139

  pGroupResInfo->pRows = NULL;

×

140

  pGroupResInfo->index = 0;

×

141

×

143

void destroyStreamFillOperatorInfo(void* param) {

×

144

  SStreamFillOperatorInfo* pInfo = (SStreamFillOperatorInfo*)param;

×

145

  destroyStreamFillInfo(pInfo->pFillInfo);

×

146

  destroyStreamFillSupporter(pInfo->pFillSup);

×

147

  blockDataDestroy(pInfo->pRes);

×

148

  pInfo->pRes = NULL;

×

149

  blockDataDestroy(pInfo->pSrcBlock);

×

150

  pInfo->pSrcBlock = NULL;

×

151

  blockDataDestroy(pInfo->pDelRes);

×

152

  pInfo->pDelRes = NULL;

×

153

  taosArrayDestroy(pInfo->matchInfo.pList);

×

154

  pInfo->matchInfo.pList = NULL;

×

155

  taosArrayDestroy(pInfo->pUpdated);

×

156

  clearGroupResArray(&pInfo->groupResInfo);

×

157

  taosArrayDestroy(pInfo->pCloseTs);

×

159

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

×

160

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

×

163

  if (pInfo->pState != NULL) {

×

164

    taosMemoryFreeClear(pInfo->pState);

×

166

  destroyStreamBasicInfo(&pInfo->basic);

×

167

  destroyNonBlockAggSupptor(&pInfo->nbSup);

×

169

  taosMemoryFree(pInfo);

×

170

×

172

static void resetFillWindow(SResultRowData* pRowData) {

×

173

  pRowData->key = INT64_MIN;

×

174

  taosMemoryFreeClear(pRowData->pRowVal);

×

175

×

177

static void resetPrevAndNextWindow(SStreamFillSupporter* pFillSup) {

×

178

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

×

179

    resetFillWindow(&pFillSup->cur);

×

181

    pFillSup->cur.key = INT64_MIN;

×

182

    pFillSup->cur.pRowVal = NULL;

×

184

  resetFillWindow(&pFillSup->prev);

×

185

  resetFillWindow(&pFillSup->next);

×

186

  resetFillWindow(&pFillSup->nextNext);

×

187

×

189

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

×

190

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

×

191

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

×

192

  resetPrevAndNextWindow(pFillSup);

×

194

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

×

195

  void*   curVal = NULL;

×

196

  int32_t curVLen = 0;

×

197

  bool    hasCurKey = true;

×

198

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

×

199

  if (code == TSDB_CODE_SUCCESS) {

×

200

    pFillSup->cur.key = key.ts;

×

201

    pFillSup->cur.pRowVal = curVal;

×

203

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

×

204

    pFillSup->cur.key = ts;

×

205

    pFillSup->cur.pRowVal = NULL;

×

206

    hasCurKey = false;

×

209

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

×

210

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

×

211

  void*            preVal = NULL;

×

212

  int32_t          preVLen = 0;

×

213

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

×

215

  if (code == TSDB_CODE_SUCCESS) {

×

216

    pFillSup->prev.key = preKey.ts;

×

217

    pFillSup->prev.pRowVal = preVal;

×

219

    if (hasCurKey) {

×

220

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

×

223

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

×

225

    pAPI->stateStore.streamStateFreeCur(pCur);

×

226

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

×

229

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

×

230

  void*   nextVal = NULL;

×

231

  int32_t nextVLen = 0;

×

232

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

×

233

  if (code == TSDB_CODE_SUCCESS) {

×

234

    pFillSup->next.key = nextKey.ts;

×

235

    pFillSup->next.pRowVal = nextVal;

×

236

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

×

237

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

×

238

      SWinKey nextNextKey = {.groupId = groupId};

×

239

      void*   nextNextVal = NULL;

×

240

      int32_t nextNextVLen = 0;

×

241

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

×

242

      if (code == TSDB_CODE_SUCCESS) {

×

243

        pFillSup->nextNext.key = nextNextKey.ts;

×

244

        pFillSup->nextNext.pRowVal = nextNextVal;

×

248

  pAPI->stateStore.streamStateFreeCur(pCur);

×

249

×

251

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

×

252

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

×

253

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

×

254

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

×

256

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

×

257

  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

×

258

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

×

259

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

×

260

    SResultCellData* pCell = getResultCell(pRowVal, i);

×

261

    if (!colDataIsNull_s(pColData, rowId)) {

×

262

      pCell->isNull = false;

×

263

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

×

264

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

×

265

      char* val = colDataGetData(pColData, rowId);

×

266

      if (IS_VAR_DATA_TYPE(pCell->type)) {

×

267

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

×

269

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

×

272

      pCell->isNull = true;

×

275

  pRowVal->key = ts;

×

276

×

278

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

×

279

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

×

280

    if (!pFillCol[i].notFillCol) {

×

281

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol + i);

×

282

      SResultCellData* pECell = getResultCell(pEndRow, slotId);

×

283

      SPoint*          pPoint = taosArrayGet(pEndPoins, slotId);

×

284

      pPoint->key = pEndRow->key;

×

285

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

×

288

×

290

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

×

291

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

×

292

  pFillInfo->start = ts;

×

293

×

295

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

×

296

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

×

297

  pFillInfo->end = ts;

×

298

×

300

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

×

301

  setFillInfoStart(start, pInterval, pFillInfo);

×

302

  pFillInfo->current = pFillInfo->start;

×

303

  setFillInfoEnd(end, pInterval, pFillInfo);

×

304

×

306

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

×

307

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

×

308

    pFillInfo->needFill = false;

×

309

    return;

×

312

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

×

313

                                pFillSup->interval.precision, NULL);

×

315

  pFillInfo->needFill = true;

×

316

  pFillInfo->start = realStart;

×

317

  pFillInfo->current = pFillInfo->start;

×

318

  pFillInfo->end = end;

×

319

  pFillInfo->pos = FILL_POS_INVALID;

×

320

  switch (pFillInfo->type) {

×

321

    case TSDB_FILL_NULL:

×

325

      break;

×

326

    case TSDB_FILL_PREV:

×

327

      pFillInfo->pResRow = &pFillSup->prev;

×

328

      break;

×

329

    case TSDB_FILL_NEXT:

×

330

      pFillInfo->pResRow = &pFillSup->next;

×

331

      break;

×

332

    case TSDB_FILL_LINEAR: {

×

333

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

×

334

      pFillInfo->pLinearInfo->hasNext = false;

×

335

      pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

336

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

×

338

      pFillInfo->pResRow = &pFillSup->prev;

×

339

      pFillInfo->pLinearInfo->winIndex = 0;

×

340

    } break;

×

341

    default:

×

342

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

×

343

      break;

×

347

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

×

348

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

×

349

    SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

350

    int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

351

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

×

352

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

×

353

    pCell->isNull = pCurCell->isNull;

×

354

    if (!pCurCell->isNull) {

×

355

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

×

358

×

360

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

×

362

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

×

363

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

×

364

    pFillInfo->needFill = false;

×

365

    pFillInfo->pos = FILL_POS_START;

×

366

    return;

×

368

  TSKEY prevWKey = INT64_MIN;

×

369

  TSKEY nextWKey = INT64_MIN;

×

370

  if (hasPrevWindow(pFillSup)) {

×

371

    prevWKey = pFillSup->prev.key;

×

373

  if (hasNextWindow(pFillSup)) {

×

374

    nextWKey = pFillSup->next.key;

×

377

  pFillInfo->needFill = true;

×

378

  pFillInfo->pos = FILL_POS_INVALID;

×

379

  switch (pFillInfo->type) {

×

380

    case TSDB_FILL_NULL:

×

384

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

×

385

        resetFillWindow(&pFillSup->prev);

×

387

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

×

388

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

×

389

          pFillInfo->preRowKey = INT64_MIN;

×

390

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

×

391

          pFillInfo->pos = FILL_POS_END;

×

393

          pFillInfo->needFill = false;

×

394

          pFillInfo->pos = FILL_POS_START;

×

396

      } else if (hasPrevWindow(pFillSup)) {

×

397

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

×

398

        pFillInfo->pos = FILL_POS_END;

×

400

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

×

401

        pFillInfo->pos = FILL_POS_START;

×

403

      copyNotFillExpData(pFillSup, pFillInfo);

×

404

    } break;

×

405

    case TSDB_FILL_PREV: {

×

406

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

×

407

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

×

408

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

×

409

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

×

410

        pFillInfo->pos = FILL_POS_START;

×

411

        resetFillWindow(&pFillSup->prev);

×

412

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

×

413

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

×

414

      } else if (hasPrevWindow(pFillSup)) {

×

415

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

×

416

        pFillInfo->pos = FILL_POS_END;

×

417

        pFillInfo->preRowKey = INT64_MIN;

×

419

      pFillInfo->pResRow = &pFillSup->prev;

×

420

    } break;

×

421

    case TSDB_FILL_NEXT: {

×

422

      if (hasPrevWindow(pFillSup)) {

×

423

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

×

424

        pFillInfo->pos = FILL_POS_END;

×

425

        resetFillWindow(&pFillSup->next);

×

426

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

×

427

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

×

428

        pFillInfo->preRowKey = INT64_MIN;

×

430

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

×

431

        pFillInfo->pos = FILL_POS_START;

×

433

      pFillInfo->pResRow = &pFillSup->next;

×

434

    } break;

×

435

    case TSDB_FILL_LINEAR: {

×

436

      pFillInfo->pLinearInfo->winIndex = 0;

×

437

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

×

438

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

×

439

        pFillInfo->pos = FILL_POS_MID;

×

440

        pFillInfo->pLinearInfo->nextEnd = nextWKey;

×

441

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

×

443

        pFillInfo->pResRow = &pFillSup->prev;

×

445

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

×

447

        pFillInfo->pLinearInfo->hasNext = true;

×

448

      } else if (hasPrevWindow(pFillSup)) {

×

449

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

×

450

        pFillInfo->pos = FILL_POS_END;

×

451

        pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

452

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

×

454

        pFillInfo->pResRow = &pFillSup->prev;

×

455

        pFillInfo->pLinearInfo->hasNext = false;

×

457

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

×

458

        pFillInfo->pos = FILL_POS_START;

×

459

        pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

460

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

×

462

        pFillInfo->pResRow = &pFillSup->cur;

×

463

        pFillInfo->pLinearInfo->hasNext = false;

×

465

    } break;

×

466

    default:

×

467

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

×

468

      break;

×

472

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

×

473

  int32_t code = TSDB_CODE_SUCCESS;

×

474

  int32_t lino = 0;

×

475

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

×

476

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

×

477

    (*pRes) = false;

×

478

    goto _end;

×

480

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

×

481

  QUERY_CHECK_CODE(code, lino, _end);

×

482

  (*pRes) = true;

×

484

_end:

×

485

  if (code != TSDB_CODE_SUCCESS) {

×

486

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

×

488

  return code;

×

491

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

×

493

  int32_t code = TSDB_CODE_SUCCESS;

×

494

  int32_t lino = 0;

×

495

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

×

496

    (*pRes) = false;

×

497

    goto _end;

×

499

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

×

500

  bool     ckRes = true;

×

501

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

×

502

  QUERY_CHECK_CODE(code, lino, _end);

×

504

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

×

505

    (*pRes) = true;

×

506

    goto _end;

×

508

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

×

509

    SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

510

    int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

511

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

×

512

    SFillInfo        tmpInfo = {

×

518

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

×

519

    if (!filled) {

×

520

      SResultCellData* pCell = getResultCell(pResRow, slotId);

×

521

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

×

522

      QUERY_CHECK_CODE(code, lino, _end);

×

525

  pBlock->info.rows++;

×

526

  (*pRes) = true;

×

528

_end:

×

529

  if (code != TSDB_CODE_SUCCESS) {

×

530

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

×

532

  return code;

×

535

bool hasRemainCalc(SStreamFillInfo* pFillInfo) {

×

536

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

×

537

    return true;

×

539

  return false;

×

542

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

×

543

  int32_t code = TSDB_CODE_SUCCESS;

×

544

  int32_t lino = 0;

×

545

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

×

546

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

×

547

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

×

548

      bool res = true;

×

549

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

×

550

      QUERY_CHECK_CODE(code, lino, _end);

×

552

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

×

553

                                     pFillSup->interval.precision, NULL);

×

556

_end:

×

557

  if (code != TSDB_CODE_SUCCESS) {

×

558

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

×

560

×

562

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

×

563

  int32_t code = TSDB_CODE_SUCCESS;

×

564

  int32_t lino = 0;

×

565

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

×

566

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

×

567

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

×

568

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

×

569

    bool        ckRes = true;

×

570

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

×

571

    QUERY_CHECK_CODE(code, lino, _end);

×

573

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

×

574

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

×

575

                                       pFillSup->interval.precision, NULL);

×

576

      pFillInfo->pLinearInfo->winIndex++;

×

577

      continue;

×

579

    pFillInfo->pLinearInfo->winIndex++;

×

580

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

×

581

      SFillColInfo* pFillCol = pFillSup->pAllColInfo + i;

×

582

      SFillInfo     tmp = {

×

583

              .currentKey = pFillInfo->current,

×

589

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

590

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

×

591

      int16_t          type = pColData->info.type;

×

592

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

×

593

      int32_t          index = pBlock->info.rows;

×

594

      if (pFillCol->notFillCol) {

×

595

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

×

596

        if (!filled) {

×

597

          code = setRowCell(pColData, index, pCell);

×

598

          QUERY_CHECK_CODE(code, lino, _end);

×

601

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

×

602

          colDataSetNULL(pColData, index);

×

603

          continue;

×

605

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

×

606

        double  vCell = 0;

×

607

        SPoint  start = {0};

×

608

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

×

609

        start.val = pCell->pData;

×

611

        SPoint cur = {0};

×

612

        cur.key = pFillInfo->current;

×

613

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

×

614

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

×

615

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

×

616

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

×

617

        QUERY_CHECK_CODE(code, lino, _end);

×

618

        destroySPoint(&cur);

×

621

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

×

622

                                     pFillSup->interval.precision, NULL);

×

623

    pBlock->info.rows++;

×

626

_end:

×

627

  if (code != TSDB_CODE_SUCCESS) {

×

628

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

×

630

×

632

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

×

633

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

×

635

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

×

636

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

×

637

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

×

638

  if (code != TSDB_CODE_SUCCESS) {

×

639

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

×

641

×

643

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

×

644

  int32_t code = TSDB_CODE_SUCCESS;

×

645

  int32_t lino = 0;

×

646

  bool    res = false;

×

647

  if (pFillInfo->needFill == false) {

×

648

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

×

649

    QUERY_CHECK_CODE(code, lino, _end);

×

650

    return;

×

653

  if (pFillInfo->pos == FILL_POS_START) {

×

654

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

×

655

    QUERY_CHECK_CODE(code, lino, _end);

×

656

    if (res) {

×

657

      pFillInfo->pos = FILL_POS_INVALID;

×

660

  if (pFillInfo->type != TSDB_FILL_LINEAR) {

×

661

    doStreamFillNormal(pFillSup, pFillInfo, pRes);

×

663

    doStreamFillLinear(pFillSup, pFillInfo, pRes);

×

665

    if (pFillInfo->pos == FILL_POS_MID) {

×

666

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

×

667

      QUERY_CHECK_CODE(code, lino, _end);

×

668

      if (res) {

×

669

        pFillInfo->pos = FILL_POS_INVALID;

×

673

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

×

674

      pFillInfo->pLinearInfo->hasNext = false;

×

675

      pFillInfo->pLinearInfo->winIndex = 0;

×

676

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

×

677

      pFillInfo->pResRow = &pFillSup->cur;

×

678

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

×

679

      doStreamFillLinear(pFillSup, pFillInfo, pRes);

×

682

  if (pFillInfo->pos == FILL_POS_END) {

×

683

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

×

684

    QUERY_CHECK_CODE(code, lino, _end);

×

685

    if (res) {

×

686

      pFillInfo->pos = FILL_POS_INVALID;

×

690

_end:

×

691

  if (code != TSDB_CODE_SUCCESS) {

×

692

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

×

696

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

×

698

  int32_t code = TSDB_CODE_SUCCESS;

×

699

  int32_t lino = 0;

×

700

  TSKEY ts = tsCol[rowId];

×

701

  pFillInfo->nextRowKey = ts;

×

702

  SResultRowData tmpNextRow = {.key = ts};

×

703

  tmpNextRow.pRowVal = taosMemoryCalloc(1, rowSize);

×

704

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

×

705

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

×

706

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

×

707

  taosMemoryFreeClear(tmpNextRow.pRowVal);

×

709

_end:

×

710

  if (code != TSDB_CODE_SUCCESS) {

×

711

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

×

713

  return code;

×

716

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

×

718

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

×

719

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

×

720

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

×

721

    resetFillWindow(&pFillSup->prev);

×

723

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

×

724

  doStreamFillRange(pFillInfo, pFillSup, pRes);

×

725

×

727

static void doStreamFillImpl(SOperatorInfo* pOperator) {

×

728

  int32_t                  code = TSDB_CODE_SUCCESS;

×

729

  int32_t                  lino = 0;

×

730

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

731

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

732

  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;

×

733

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

734

  SSDataBlock*             pBlock = pInfo->pSrcBlock;

×

735

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

×

736

  SSDataBlock*             pRes = pInfo->pRes;

×

737

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

×

738

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

×

739

  pRes->info.id.groupId = groupId;

×

740

  pInfo->srcRowIndex++;

×

742

  if (pInfo->srcRowIndex == 0) {

×

743

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

×

744

    QUERY_CHECK_CODE(code, lino, _end);

×

745

    pInfo->srcRowIndex++;

×

748

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

×

749

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

×

750

    QUERY_CHECK_CODE(code, lino, _end);

×

751

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

×

752

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

×

753

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

×

754

      QUERY_CHECK_CODE(code, lino, _end);

×

755

      return;

×

757

    pInfo->srcRowIndex++;

×

759

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

×

760

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

×

761

  QUERY_CHECK_CODE(code, lino, _end);

×

762

  blockDataCleanup(pInfo->pSrcBlock);

×

764

_end:

×

765

  if (code != TSDB_CODE_SUCCESS) {

×

766

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

×

770

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

×

771

  int32_t          code = TSDB_CODE_SUCCESS;

×

772

  int32_t          lino = 0;

×

773

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

×

774

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

×

775

  SExecTaskInfo*   pTaskInfo = pOp->pTaskInfo;

×

776

  SSDataBlock*     pBlock = delRes;

×

777

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

×

778

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

×

779

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

×

780

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

×

781

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

×

782

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

×

783

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

×

784

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

×

785

  QUERY_CHECK_CODE(code, lino, _end);

×

787

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

×

788

  QUERY_CHECK_CODE(code, lino, _end);

×

790

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

×

791

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

×

792

  QUERY_CHECK_CODE(code, lino, _end);

×

794

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

×

795

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

×

797

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

×

799

  void*   tbname = NULL;

×

800

  int32_t winCode = TSDB_CODE_SUCCESS;

×

801

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

×

802

  QUERY_CHECK_CODE(code, lino, _end);

×

803

  if (winCode != TSDB_CODE_SUCCESS) {

×

804

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

×

807

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

×

808

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

×

809

    QUERY_CHECK_CODE(code, lino, _end);

×

810

    pAPI->stateStore.streamStateFreeVal(tbname);

×

813

  pBlock->info.rows++;

×

815

_end:

×

816

  if (code != TSDB_CODE_SUCCESS) {

×

817

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

×

819

  return code;

×

822

int32_t buildDeleteResult(SOperatorInfo* pOperator, TSKEY startTs, TSKEY endTs, uint64_t groupId, SSDataBlock* delRes) {

×

823

  int32_t                  code = TSDB_CODE_SUCCESS;

×

824

  int32_t                  lino = 0;

×

825

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

826

  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;

×

827

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

828

  if (hasPrevWindow(pFillSup)) {

×

829

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

×

830

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

×

831

    QUERY_CHECK_CODE(code, lino, _end);

×

832

  } else if (hasNextWindow(pFillSup)) {

×

833

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

×

834

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

×

835

    QUERY_CHECK_CODE(code, lino, _end);

×

837

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

×

838

    QUERY_CHECK_CODE(code, lino, _end);

×

841

_end:

×

842

  if (code != TSDB_CODE_SUCCESS) {

×

843

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

×

845

  return code;

×

848

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

×

849

  int32_t                  code = TSDB_CODE_SUCCESS;

×

850

  int32_t                  lino = 0;

×

851

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

×

852

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

853

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

854

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

×

855

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

×

856

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

×

857

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

×

858

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

×

859

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

×

860

    QUERY_CHECK_CODE(code, lino, _end);

×

862

    STimeFillRange tw = {

×

867

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

×

868

    if (!tmp) {

×

869

      code = terrno;

×

870

      QUERY_CHECK_CODE(code, lino, _end);

×

874

_end:

×

875

  if (code != TSDB_CODE_SUCCESS) {

×

876

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

×

878

  return code;

×

881

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

×

882

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

×

883

  void*   val = NULL;

×

884

  int32_t len = 0;

×

885

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

×

886

  if (code != TSDB_CODE_SUCCESS) {

×

887

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

×

889

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

×

890

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

×

891

    pAPI->stateStore.streamStateFreeCur(pCur);

×

892

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

×

895

  if (code == TSDB_CODE_SUCCESS) {

×

896

    resetFillWindow(pWinData);

×

897

    pWinData->key = key.ts;

×

898

    pWinData->pRowVal = val;

×

900

×

902

static void doDeleteFillFinalize(SOperatorInfo* pOperator) {

×

903

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

×

905

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

906

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

907

  int32_t                  size = taosArrayGetSize(pFillInfo->delRanges);

×

908

  while (pFillInfo->delIndex < size) {

×

909

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

×

910

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

×

911

      return;

×

913

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

×

914

    TSKEY realEnd = range->ekey + 1;

×

915

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

×

916

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

×

917

                         &pInfo->pFillSup->next);

×

919

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

×

920

    pFillInfo->delIndex++;

×

921

    if (pInfo->pFillInfo->needFill) {

×

922

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

×

923

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

×

928

static int32_t doDeleteFillResult(SOperatorInfo* pOperator) {

×

929

  int32_t                  code = TSDB_CODE_SUCCESS;

×

930

  int32_t                  lino = 0;

×

931

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

×

932

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

933

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

934

  SSDataBlock*             pBlock = pInfo->pSrcDelBlock;

×

935

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

937

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

×

938

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

×

939

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

×

940

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

×

941

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

×

942

    TSKEY            ts = tsStarts[pInfo->srcDelRowIndex];

×

943

    TSKEY            endTs = ts;

×

944

    uint64_t         groupId = groupIds[pInfo->srcDelRowIndex];

×

945

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

×

946

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

×

948

    if (!pCur) {

×

949

      pInfo->srcDelRowIndex++;

×

950

      continue;

×

953

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

×

954

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

×

955

      TSKEY    delTs = tsStarts[pInfo->srcDelRowIndex];

×

956

      uint64_t delGroupId = groupIds[pInfo->srcDelRowIndex];

×

957

      int32_t  winCode = TSDB_CODE_SUCCESS;

×

958

      if (groupId != delGroupId) {

×

959

        break;

×

961

      if (delTs > nextKey.ts) {

×

962

        break;

×

965

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

×

966

      if (delTs == nextKey.ts) {

×

967

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

×

968

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

×

970

        if (delTs != ts) {

×

971

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

×

972

          pAPI->stateStore.streamStateFreeCur(pCur);

×

973

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

×

975

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

×

976

        if (winCode != TSDB_CODE_SUCCESS) {

×

977

          break;

×

980

      pInfo->srcDelRowIndex++;

×

983

    pAPI->stateStore.streamStateFreeCur(pCur);

×

984

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

×

985

    QUERY_CHECK_CODE(code, lino, _end);

×

988

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

×

990

_end:

×

991

  if (code != TSDB_CODE_SUCCESS) {

×

992

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

×

994

  return code;

×

997

void resetStreamFillSup(SStreamFillSupporter* pFillSup) {

×

998

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

999

  SSHashObj* pNewMap = tSimpleHashInit(16, hashFn);

×

1000

  if (pNewMap != NULL) {

×

1001

    tSimpleHashCleanup(pFillSup->pResMap);

×

1002

    pFillSup->pResMap = pNewMap;

×

1004

    tSimpleHashClear(pFillSup->pResMap);

×

1006

  pFillSup->hasDelete = false;

×

1007

×

1008

void resetStreamFillInfo(SStreamFillOperatorInfo* pInfo) {

×

1009

  resetStreamFillSup(pInfo->pFillSup);

×

1010

  taosArrayClear(pInfo->pFillInfo->delRanges);

×

1011

  pInfo->pFillInfo->delIndex = 0;

×

1012

×

1014

int32_t doApplyStreamScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pSrcBlock,

×

1016

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1017

  int32_t                  lino = 0;

×

1018

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1019

  SExprSupp*               pSup = &pOperator->exprSupp;

×

1020

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1022

  blockDataCleanup(pDstBlock);

×

1023

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

×

1024

  QUERY_CHECK_CODE(code, lino, _end);

×

1026

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

×

1027

  QUERY_CHECK_CODE(code, lino, _end);

×

1028

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

×

1029

  QUERY_CHECK_CODE(code, lino, _end);

×

1031

  pDstBlock->info.rows = 0;

×

1032

  pSup = &pInfo->pFillSup->notFillExprSup;

×

1033

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

×

1034

  QUERY_CHECK_CODE(code, lino, _end);

×

1035

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

×

1036

  QUERY_CHECK_CODE(code, lino, _end);

×

1038

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

×

1040

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

×

1042

_end:

×

1043

  if (code != TSDB_CODE_SUCCESS) {

×

1044

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

×

1046

  return code;

×

1049

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

×

1050

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1051

  int32_t                  lino = 0;

×

1052

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1053

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1055

  if (pOperator->status == OP_EXEC_DONE) {

×

1056

    (*ppRes) = NULL;

×

1057

    return code;

×

1059

  blockDataCleanup(pInfo->pRes);

×

1060

  if (hasRemainCalc(pInfo->pFillInfo) ||

×

1061

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

×

1062

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

×

1063

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

×

1064

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

×

1065

      (*ppRes) = pInfo->pRes;

×

1066

      return code;

×

1069

  if (pOperator->status == OP_RES_TO_RETURN) {

×

1070

    doDeleteFillFinalize(pOperator);

×

1071

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

×

1072

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

×

1073

      (*ppRes) = pInfo->pRes;

×

1074

      return code;

×

1076

    setOperatorCompleted(pOperator);

×

1077

    resetStreamFillInfo(pInfo);

×

1078

    (*ppRes) = NULL;

×

1079

    return code;

×

1082

  SSDataBlock*   fillResult = NULL;

×

1083

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

×

1085

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

×

1087

      SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

1088

      if (pBlock == NULL) {

×

1089

        pOperator->status = OP_RES_TO_RETURN;

×

1090

        pInfo->pFillInfo->preRowKey = INT64_MIN;

×

1091

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

×

1092

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

×

1093

          (*ppRes) = pInfo->pRes;

×

1094

          return code;

×

1096

        break;

×

1098

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

×

1100

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

×

1101

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

×

1102

        pInfo->pFillInfo->preRowKey = INT64_MIN;

×

1105

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

×

1106

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

×

1107

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

×

1110

      switch (pBlock->info.type) {

×

1111

        case STREAM_RETRIEVE:

×

1112

          (*ppRes) = pBlock;

×

1113

          return code;

×

1114

        case STREAM_DELETE_RESULT: {

×

1115

          pInfo->pSrcDelBlock = pBlock;

×

1116

          pInfo->srcDelRowIndex = 0;

×

1117

          blockDataCleanup(pInfo->pDelRes);

×

1118

          pInfo->pFillSup->hasDelete = true;

×

1119

          code = doDeleteFillResult(pOperator);

×

1120

          QUERY_CHECK_CODE(code, lino, _end);

×

1122

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

×

1123

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

×

1124

            (*ppRes) = pInfo->pDelRes;

×

1125

            return code;

×

1127

          continue;

×

1129

        case STREAM_NORMAL:

×

1132

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

×

1133

          QUERY_CHECK_CODE(code, lino, _end);

×

1135

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

×

1136

          pInfo->srcRowIndex = -1;

×

1137

        } break;

×

1138

        case STREAM_CHECKPOINT:

×

1140

          (*ppRes) = pBlock;

×

1141

          return code;

×

1143

        default:

×

1144

          return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1148

    doStreamFillImpl(pOperator);

×

1149

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

×

1150

    QUERY_CHECK_CODE(code, lino, _end);

×

1152

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

×

1153

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

×

1154

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

×

1155

      break;

×

1158

  if (pOperator->status == OP_RES_TO_RETURN) {

×

1159

    doDeleteFillFinalize(pOperator);

×

1162

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

×

1163

    setOperatorCompleted(pOperator);

×

1164

    resetStreamFillInfo(pInfo);

×

1165

    (*ppRes) = NULL;

×

1166

    return code;

×

1169

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

×

1170

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

×

1171

  (*ppRes) = pInfo->pRes;

×

1172

  return code;

×

1174

_end:

×

1175

  if (code != TSDB_CODE_SUCCESS) {

×

1176

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

×

1177

    pTaskInfo->code = code;

×

1178

    T_LONG_JMP(pTaskInfo->env, code);

×

1180

  setOperatorCompleted(pOperator);

×

1181

  resetStreamFillInfo(pInfo);

×

1182

  (*ppRes) = NULL;

×

1183

  return code;

×

1186

static void resetForceFillWindow(SResultRowData* pRowData) {

×

1187

  pRowData->key = INT64_MIN;

×

1188

  pRowData->pRowVal = NULL;

×

1189

×

1191

void doBuildForceFillResultImpl(SOperatorInfo* pOperator, SStreamFillSupporter* pFillSup,

×

1193

  int32_t code = TSDB_CODE_SUCCESS;

×

1194

  int32_t lino = 0;

×

1196

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1197

  bool                     res = false;

×

1198

  int32_t                  numOfRows = getNumOfTotalRes(pGroupResInfo);

×

1199

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

×

1200

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

×

1201

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

×

1202

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

×

1203

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

×

1204

      break;

×

1207

    SRowBuffPos* pValPos = NULL;

×

1208

    int32_t      len = 0;

×

1209

    int32_t      winCode = TSDB_CODE_SUCCESS;

×

1210

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

×

1211

    QUERY_CHECK_CODE(code, lino, _end);

×

1212

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

×

1213

    if (winCode == TSDB_CODE_SUCCESS) {

×

1214

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

×

1215

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

×

1216

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

×

1217

      QUERY_CHECK_CODE(code, lino, _end);

×

1218

      resetForceFillWindow(&pFillSup->cur);

×

1219

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

×

1221

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

×

1222

      SRowBuffPos* prePos = NULL;

×

1223

      int32_t      preVLen = 0;

×

1224

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

×

1226

      QUERY_CHECK_CODE(code, lino, _end);

×

1227

      if (winCode == TSDB_CODE_SUCCESS) {

×

1228

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

×

1229

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

×

1230

        if (pFillInfo->type == TSDB_FILL_PREV) {

×

1231

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

×

1232

          QUERY_CHECK_CODE(code, lino, _end);

×

1234

          copyNotFillExpData(pFillSup, pFillInfo);

×

1235

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

×

1236

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

×

1237

          QUERY_CHECK_CODE(code, lino, _end);

×

1239

        resetForceFillWindow(&pFillSup->cur);

×

1241

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

×

1245

_end:

×

1246

  if (code != TSDB_CODE_SUCCESS) {

×

1247

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

×

1249

×

1251

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

×

1253

  blockDataCleanup(pBlock);

×

1254

  if (!hasRemainResults(pGroupResInfo)) {

×

1255

    return;

×

1259

  pBlock->info.id.groupId = 0;

×

1260

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

×

1263

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

×

1264

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1265

  int32_t                  lino = 0;

×

1266

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1267

  uint16_t                 opType = pOperator->operatorType;

×

1268

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1270

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

×

1271

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

×

1272

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

×

1273

    (*ppRes) = pInfo->pRes;

×

1274

    goto _end;

×

1277

  (*ppRes) = NULL;

×

1279

_end:

×

1280

  if (code != TSDB_CODE_SUCCESS) {

×

1281

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

×

1283

  return code;

×

1286

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

×

1287

  int32_t code = TSDB_CODE_SUCCESS;

×

1288

  int32_t lino = 0;

×

1290

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

×

1291

  int32_t      curVLen = 0;

×

1292

  SRowBuffPos* pStatePos = NULL;

×

1293

  int32_t      winCode = TSDB_CODE_SUCCESS;

×

1294

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

×

1296

  QUERY_CHECK_CODE(code, lino, _end);

×

1297

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

×

1298

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

×

1300

_end:

×

1301

  if (code != TSDB_CODE_SUCCESS) {

×

1302

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

×

1304

×

1306

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

×

1308

  int32_t code = TSDB_CODE_SUCCESS;

×

1309

  int32_t lino = 0;

×

1310

  TSKEY ts = tsCol[rowId];

×

1311

  pFillInfo->nextRowKey = ts;

×

1312

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

×

1313

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

×

1315

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

×

1316

  keepResultInStateBuf(pInfo, groupId, &tmpNextRow);

×

1318

_end:

×

1319

  if (code != TSDB_CODE_SUCCESS) {

×

1320

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

×

1322

  return code;

×

1326

static int32_t doStreamForceFillImpl(SOperatorInfo* pOperator) {

×

1327

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1328

  int32_t                  lino = 0;

×

1329

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1330

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1331

  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;

×

1332

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

1333

  SSDataBlock*             pBlock = pInfo->pSrcBlock;

×

1334

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

×

1335

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

×

1336

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

×

1337

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

×

1338

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

×

1339

    QUERY_CHECK_CODE(code, lino, _end);

×

1341

    int32_t size =  taosArrayGetSize(pInfo->pCloseTs);

×

1342

    if (size > 0) {

×

1343

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

×

1344

      TSKEY  resTs = tsCol[i];

×

1345

      while (resTs < (*pTs)) {

×

1346

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

×

1347

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

×

1348

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

×

1350

        if (IS_FILL_CONST_VALUE(pFillSup->type)) {

×

1353

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

×

1354

                            pFillSup->interval.precision, NULL);

×

1358

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

×

1359

  QUERY_CHECK_CODE(code, lino, _end);

×

1361

_end:

×

1362

  if (code != TSDB_CODE_SUCCESS) {

×

1363

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

×

1365

  return code;

×

1368

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

×

1369

  int32_t          code = TSDB_CODE_SUCCESS;

×

1370

  int32_t          lino = 0;

×

1371

  int64_t          groupId = 0;

×

1372

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

×

1373

  while (1) {

×

1374

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

×

1375

    if (winCode != TSDB_CODE_SUCCESS) {

×

1376

      break;

×

1378

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

×

1379

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

×

1380

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

×

1382

    pStateStore->streamStateGroupCurNext(pCur);

×

1384

  pStateStore->streamStateFreeCur(pCur);

×

1385

  pCur = NULL;

×

1387

_end:

×

1388

  if (code != TSDB_CODE_SUCCESS) {

×

1389

    pStateStore->streamStateFreeCur(pCur);

×

1390

    pCur = NULL;

×

1391

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

×

1393

  return code;

×

1396

void removeDuplicateResult(SArray* pTsArrray, __compar_fn_t fn) {

×

1397

  taosArraySort(pTsArrray, fn);

×

1398

  taosArrayRemoveDuplicate(pTsArrray, fn, NULL);

×

1399

×

1402

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

×

1403

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1404

  int32_t                  lino = 0;

×

1405

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1406

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1408

  if (pOperator->status == OP_EXEC_DONE) {

×

1409

    (*ppRes) = NULL;

×

1410

    return code;

×

1413

  if (pOperator->status == OP_RES_TO_RETURN) {

×

1414

    SSDataBlock* resBlock = NULL;

×

1415

    code = buildForceFillResult(pOperator, &resBlock);

×

1416

    QUERY_CHECK_CODE(code, lino, _end);

×

1418

    if (resBlock != NULL) {

×

1419

      (*ppRes) = resBlock;

×

1420

      goto _end;

×

1423

    pInfo->stateStore.streamStateClearExpiredState(pInfo->pState, 1, INT64_MAX);

×

1424

    resetStreamFillInfo(pInfo);

×

1425

    setStreamOperatorCompleted(pOperator);

×

1426

    (*ppRes) = NULL;

×

1427

    goto _end;

×

1430

  SSDataBlock*   fillResult = NULL;

×

1431

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

×

1432

  while (1) {

×

1433

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

1434

    if (pBlock == NULL) {

×

1435

      pOperator->status = OP_RES_TO_RETURN;

×

1436

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

×

1437

      break;

×

1439

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

×

1440

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

×

1442

    switch (pBlock->info.type) {

×

1443

      case STREAM_NORMAL:

×

1445

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

×

1446

        QUERY_CHECK_CODE(code, lino, _end);

×

1448

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

×

1449

        pInfo->srcRowIndex = -1;

×

1450

      } break;

×

1451

      case STREAM_CHECKPOINT: {

×

1452

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

×

1453

        (*ppRes) = pBlock;

×

1454

        goto _end;

×

1456

      case STREAM_CREATE_CHILD_TABLE: {

×

1457

        (*ppRes) = pBlock;

×

1458

        goto _end;

×

1460

      case STREAM_GET_RESULT: {

×

1461

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

×

1462

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

×

1463

        continue;

×

1465

      default:

×

1466

        code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1467

        QUERY_CHECK_CODE(code, lino, _end);

×

1470

    code = doStreamForceFillImpl(pOperator);

×

1471

    QUERY_CHECK_CODE(code, lino, _end);

×

1474

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

×

1475

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

×

1476

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

×

1477

    QUERY_CHECK_CODE(code, lino, _end);

×

1479

  taosArrayClear(pInfo->pCloseTs);

×

1480

  removeDuplicateResult(pInfo->pUpdated, winKeyCmprImpl);

×

1482

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

×

1483

  pInfo->groupResInfo.freeItem = false;

×

1485

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

×

1486

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

×

1488

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

×

1489

  QUERY_CHECK_CODE(code, lino, _end);

×

1491

  code = buildForceFillResult(pOperator, ppRes);

×

1492

  QUERY_CHECK_CODE(code, lino, _end);

×

1494

  if ((*ppRes) == NULL) {

×

1495

    pInfo->stateStore.streamStateClearExpiredState(pInfo->pState, 1, INT64_MAX);

×

1496

    resetStreamFillInfo(pInfo);

×

1497

    setStreamOperatorCompleted(pOperator);

×

1500

_end:

×

1501

  if (code != TSDB_CODE_SUCCESS) {

×

1502

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

×

1503

    pTaskInfo->code = code;

×

1505

  return code;

×

1508

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

×

1509

  int32_t numOfCols = taosArrayGetSize(pInputRes->pDataBlock);

×

1510

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

×

1511

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

×

1512

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

×

1513

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

×

1515

  pFillSup->next.key = INT64_MIN;

×

1516

  pFillSup->nextNext.key = INT64_MIN;

×

1517

  pFillSup->prev.key = INT64_MIN;

×

1518

  pFillSup->cur.key = INT64_MIN;

×

1519

  pFillSup->next.pRowVal = NULL;

×

1520

  pFillSup->nextNext.pRowVal = NULL;

×

1521

  pFillSup->prev.pRowVal = NULL;

×

1522

  pFillSup->cur.pRowVal = NULL;

×

1524

  return TSDB_CODE_SUCCESS;

×

1527

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

×

1529

  int32_t               code = TSDB_CODE_SUCCESS;

×

1530

  int32_t               lino = 0;

×

1531

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

×

1532

  if (!pFillSup) {

×

1533

    code = terrno;

×

1534

    QUERY_CHECK_CODE(code, lino, _end);

×

1536

  pFillSup->numOfFillCols = numOfFillCols;

×

1537

  int32_t    numOfNotFillCols = 0;

×

1538

  SExprInfo* noFillExprInfo = NULL;

×

1540

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

×

1541

  QUERY_CHECK_CODE(code, lino, _end);

×

1543

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

×

1544

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

×

1545

  if (pFillSup->pAllColInfo == NULL) {

×

1546

    code = terrno;

×

1547

    lino = __LINE__;

×

1548

    destroyExprInfo(noFillExprInfo, numOfNotFillCols);

×

1549

    goto _end;

×

1552

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

×

1553

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

×

1554

  pFillSup->interval = *pInterval;

×

1555

  pFillSup->pAPI = pAPI;

×

1557

  code = initResultBuf(pInputRes, pFillSup);

×

1558

  QUERY_CHECK_CODE(code, lino, _end);

×

1560

  SExprInfo* noFillExpr = NULL;

×

1561

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

×

1562

  QUERY_CHECK_CODE(code, lino, _end);

×

1564

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

×

1565

  QUERY_CHECK_CODE(code, lino, _end);

×

1567

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

1568

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

×

1569

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

×

1570

  pFillSup->hasDelete = false;

×

1571

  pFillSup->normalFill = true;

×

1572

  pFillSup->pResultRange = taosArrayInit(2, POINTER_BYTES);

×

1575

_end:

×

1576

  if (code != TSDB_CODE_SUCCESS) {

×

1577

    destroyStreamFillSupporter(pFillSup);

×

1578

    pFillSup = NULL;

×

1579

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

×

1581

  return pFillSup;

×

1584

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

×

1585

  int32_t          code = TSDB_CODE_SUCCESS;

×

1586

  int32_t          lino = 0;

×

1587

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

×

1588

  if (!pFillInfo) {

×

1589

    code = terrno;

×

1590

    QUERY_CHECK_CODE(code, lino, _end);

×

1593

  pFillInfo->start = INT64_MIN;

×

1594

  pFillInfo->current = INT64_MIN;

×

1595

  pFillInfo->end = INT64_MIN;

×

1596

  pFillInfo->preRowKey = INT64_MIN;

×

1597

  pFillInfo->needFill = false;

×

1598

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

×

1599

  if (!pFillInfo) {

×

1600

    code = terrno;

×

1601

    QUERY_CHECK_CODE(code, lino, _end);

×

1604

  pFillInfo->pLinearInfo->hasNext = false;

×

1605

  pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

1606

  pFillInfo->pLinearInfo->pEndPoints = NULL;

×

1607

  pFillInfo->pLinearInfo->pNextEndPoints = NULL;

×

1608

  if (pFillSup->type == TSDB_FILL_LINEAR) {

×

1609

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

×

1610

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

×

1611

      code = terrno;

×

1612

      QUERY_CHECK_CODE(code, lino, _end);

×

1615

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

×

1616

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

×

1617

      code = terrno;

×

1618

      QUERY_CHECK_CODE(code, lino, _end);

×

1621

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

×

1622

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

×

1623

      if (pColData == NULL) {

×

1624

        SPoint dummy = {0};

×

1625

        dummy.val = taosMemoryCalloc(1, 1);

×

1626

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

×

1627

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

×

1629

        dummy.val = taosMemoryCalloc(1, 1);

×

1630

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

×

1631

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

×

1633

        continue;

×

1635

      SPoint value = {0};

×

1636

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

×

1637

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

×

1639

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

×

1640

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

×

1642

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

×

1643

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

×

1645

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

×

1646

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

×

1649

  pFillInfo->pLinearInfo->winIndex = 0;

×

1651

  pFillInfo->pNonFillRow = NULL;

×

1652

  pFillInfo->pResRow = NULL;

×

1653

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

×

1654

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

×

1655

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

×

1656

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

×

1658

    pFillInfo->pResRow->key = INT64_MIN;

×

1659

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

×

1660

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

×

1662

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

×

1663

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

×

1664

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

×

1665

      if (pColData == NULL) {

×

1666

        pCell->bytes = 1;

×

1667

        pCell->type = 4;

×

1668

        continue;

×

1670

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

×

1671

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

×

1674

    int32_t numOfResCol = taosArrayGetSize(pRes->pDataBlock);

×

1675

    if (numOfResCol < pFillSup->numOfAllCols) {

×

1676

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

×

1677

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

×

1678

      pFillSup->pOffsetInfo = pTmpBuf;

×

1680

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

×

1681

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

×

1682

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

×

1683

        pFillSup->pOffsetInfo[i] = preLength;

×

1684

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

×

1685

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

×

1689

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

×

1690

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

×

1691

    pFillInfo->pNonFillRow->key = INT64_MIN;

×

1692

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

×

1693

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

×

1696

  pFillInfo->type = pFillSup->type;

×

1697

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

×

1698

  if (!pFillInfo->delRanges) {

×

1699

    code = terrno;

×

1700

    QUERY_CHECK_CODE(code, lino, _end);

×

1703

  pFillInfo->delIndex = 0;

×

1704

  pFillInfo->curGroupId = 0;

×

1705

  pFillInfo->hasNext = false;

×

1706

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

×

1707

  return pFillInfo;

×

1709

_end:

×

1710

  if (code != TSDB_CODE_SUCCESS) {

×

1711

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

×

1713

  destroyStreamFillInfo(pFillInfo);

×

1714

  return NULL;

×

1717

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

×

1718

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

×

1719

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

×

1720

      SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

1721

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

1722

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

×

1723

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

×

1724

      if (pCell->type == TSDB_DATA_TYPE_FLOAT) {

×

1725

        float v = 0;

×

1726

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

×

1727

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

×

1728

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

×

1729

        double v = 0;

×

1730

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

×

1731

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

×

1732

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

×

1733

        int64_t v = 0;

×

1734

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

×

1735

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

×

1737

        pCell->isNull = true;

×

1740

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

×

1741

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

×

1742

      SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

1743

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

1744

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

×

1745

      pCell->isNull = true;

×

1748

×

1750

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

×

1752

  int32_t code = TSDB_CODE_SUCCESS;

×

1753

  int32_t lino = 0;

×

1754

  if (IS_NORMAL_INTERVAL_OP(downstream)) {

×

1755

    SStreamIntervalOperatorInfo* pInfo = downstream->info;

×

1756

    *triggerType = pInfo->twAggSup.calTrigger;

×

1757

    *pInterval = pInfo->interval;

×

1758

    *pOperatorFlag = pInfo->basic.operatorFlag;

×

1760

    SStreamIntervalSliceOperatorInfo* pInfo = downstream->info;

×

1761

    *triggerType = pInfo->twAggSup.calTrigger;

×

1762

    *pInterval = pInfo->interval;

×

1763

    pInfo->hasFill = true;

×

1764

    *pOperatorFlag = pInfo->basic.operatorFlag;

×

1767

  QUERY_CHECK_CODE(code, lino, _end);

×

1769

_end:

×

1770

  if (code != TSDB_CODE_SUCCESS) {

×

1771

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

×

1773

  return code;

×

1776

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

×

1778

  int32_t code = TSDB_CODE_SUCCESS;

×

1779

  int32_t lino = 0;

×

1781

  pInfo->stateStore = *pStore;

×

1782

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

×

1783

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

×

1785

  *(pInfo->pState) = *pState;

×

1786

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

×

1787

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

×

1788

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

×

1789

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

×

1790

  QUERY_CHECK_CODE(code, lino, _end);

×

1792

_end:

×

1793

  if (code != TSDB_CODE_SUCCESS) {

×

1794

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

×

1796

  return code;

×

1799

int32_t createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode,

×

1801

  QRY_PARAM_CHECK(pOptrInfo);

×

1803

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1804

  int32_t                  lino = 0;

×

1805

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

×

1806

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

×

1807

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

×

1808

    code = terrno;

×

1809

    QUERY_CHECK_CODE(code, lino, _error);

×

1812

  int32_t    numOfFillCols = 0;

×

1813

  SExprInfo* pFillExprInfo = NULL;

×

1815

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

×

1816

  QUERY_CHECK_CODE(code, lino, _error);

×

1818

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

×

1819

  QUERY_CHECK_CODE(code, lino, _error);

×

1821

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

×

1822

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

×

1824

  int8_t triggerType = 0;

×

1825

  SInterval interval = {0};

×

1826

  int16_t opFlag = 0;

×

1827

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

×

1828

  QUERY_CHECK_CODE(code, lino, _error);

×

1830

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

×

1832

  if (!pInfo->pFillSup) {

×

1833

    code = TSDB_CODE_FAILED;

×

1834

    QUERY_CHECK_CODE(code, lino, _error);

×

1837

  initResultSizeInfo(&pOperator->resultInfo, 4096);

×

1838

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

×

1839

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

×

1841

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

×

1842

  QUERY_CHECK_CODE(code, lino, _error);

×

1844

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

×

1845

  QUERY_CHECK_CODE(code, lino, _error);

×

1847

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

×

1848

  if (!pInfo->pFillInfo) {

×

1849

    goto _error;

×

1852

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

×

1854

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

×

1855

  QUERY_CHECK_CODE(code, lino, _error);

×

1857

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

×

1858

  QUERY_CHECK_CODE(code, lino, _error);

×

1860

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

×

1861

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

×

1863

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

×

1864

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

×

1866

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

×

1867

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

×

1869

  int32_t numOfOutputCols = 0;

×

1870

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

×

1872

  QUERY_CHECK_CODE(code, lino, _error);

×

1874

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

×

1875

  QUERY_CHECK_CODE(code, lino, _error);

×

1877

  pInfo->srcRowIndex = -1;

×

1878

  setOperatorInfo(pOperator, "StreamFillOperator", nodeType(pPhyFillNode), false, OP_NOT_OPENED, pInfo,

×

1881

  if (triggerType == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {

×

1882

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

×

1883

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

×

1884

    QUERY_CHECK_CODE(code, lino, _error);

×

1885

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

×

1887

  } else if (triggerType == STREAM_TRIGGER_CONTINUOUS_WINDOW_CLOSE) {

×

1888

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

×

1889

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

×

1890

    QUERY_CHECK_CODE(code, lino, _error);

×

1892

    initNonBlockAggSupptor(&pInfo->nbSup, &pInfo->pFillSup->interval, downstream);

×

1893

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

×

1894

    QUERY_CHECK_CODE(code, lino, _error);

×

1896

    code = streamClientCheckCfg(&pInfo->nbSup.recParam);

×

1897

    QUERY_CHECK_CODE(code, lino, _error);

×

1899

    pInfo->basic.operatorFlag = opFlag;

×

1900

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

×

1901

      pInfo->nbSup.numOfKeep++;

×

1903

    code = initFillSupRowInfo(pInfo->pFillSup, pInfo->pRes);

×

1904

    QUERY_CHECK_CODE(code, lino, _error);

×

1905

    pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamNonblockFillNext, NULL, destroyStreamNonblockFillOperatorInfo,

×

1908

    pInfo->pState = NULL;

×

1909

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

×

1912

  setOperatorStreamStateFn(pOperator, streamOpReleaseState, streamOpReloadState);

×

1914

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

×

1915

  QUERY_CHECK_CODE(code, lino, _error);

×

1917

  *pOptrInfo = pOperator;

×

1918

  return TSDB_CODE_SUCCESS;

×

1920

_error:

×

1921

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

×

1923

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

×

1924

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);

×

1925

  pTaskInfo->code = code;

×

1926

  return code;

×

taosdata / TDengine / #3796

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

taosdata / TDengine / #3796

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