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

47

TSKEY getNextWindowTs(TSKEY ts, SInterval* pInterval) {

×

48

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

×

49

  getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);

×

50

  return win.skey;

×

53

TSKEY getPrevWindowTs(TSKEY ts, SInterval* pInterval) {

×

54

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

×

55

  getNextTimeWindow(pInterval, &win, TSDB_ORDER_DESC);

×

56

  return win.skey;

×

59

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

×

60

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

×

63

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

×

64

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

×

65

    return NULL;

×

67

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

×

68

  SResultCellData* pCell = pRaw->pRowVal;

×

69

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

×

70

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

×

71

    pCell = (SResultCellData*)pData;

×

73

  return pCell;

×

76

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

×

77

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

×

78

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

×

79

    taosVariantDestroy(&pFillCol[i].fillVal);

×

81

  if (start < end) {

×

82

    taosMemoryFreeClear(pFillCol[start].pExpr);

×

84

  taosMemoryFree(pFillCol);

×

85

  return NULL;

×

88

void destroyStreamFillSupporter(SStreamFillSupporter* pFillSup) {

×

89

  if (pFillSup == NULL) {

×

90

    return;

×

92

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

×

93

  tSimpleHashCleanup(pFillSup->pResMap);

×

94

  pFillSup->pResMap = NULL;

×

95

  cleanupExprSupp(&pFillSup->notFillExprSup);

×

96

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

×

97

    taosMemoryFree(pFillSup->cur.pRowVal);

×

99

  taosMemoryFree(pFillSup->prev.pRowVal);

×

100

  taosMemoryFree(pFillSup->next.pRowVal);

×

101

  taosMemoryFree(pFillSup->nextNext.pRowVal);

×

103

  taosMemoryFree(pFillSup->pOffsetInfo);

×

105

  taosMemoryFree(pFillSup);

×

108

void destroySPoint(void* ptr) {

×

109

  SPoint* point = (SPoint*)ptr;

×

110

  taosMemoryFreeClear(point->val);

×

111

×

113

void destroyStreamFillLinearInfo(SStreamFillLinearInfo* pFillLinear) {

×

114

  taosArrayDestroyEx(pFillLinear->pEndPoints, destroySPoint);

×

115

  taosArrayDestroyEx(pFillLinear->pNextEndPoints, destroySPoint);

×

116

  taosMemoryFree(pFillLinear);

×

117

×

119

void destroyStreamFillInfo(SStreamFillInfo* pFillInfo) {

×

120

  if (pFillInfo == NULL) {

×

121

    return;

×

123

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

×

124

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

×

125

    taosMemoryFreeClear(pFillInfo->pResRow->pRowVal);

×

126

    taosMemoryFreeClear(pFillInfo->pResRow);

×

127

    taosMemoryFreeClear(pFillInfo->pNonFillRow->pRowVal);

×

128

    taosMemoryFreeClear(pFillInfo->pNonFillRow);

×

130

  destroyStreamFillLinearInfo(pFillInfo->pLinearInfo);

×

131

  pFillInfo->pLinearInfo = NULL;

×

133

  taosArrayDestroy(pFillInfo->delRanges);

×

134

  taosMemoryFreeClear(pFillInfo->pTempBuff);

×

135

  taosMemoryFree(pFillInfo);

×

138

static void destroyStreamFillOperatorInfo(void* param) {

×

139

  SStreamFillOperatorInfo* pInfo = (SStreamFillOperatorInfo*)param;

×

140

  destroyStreamFillInfo(pInfo->pFillInfo);

×

141

  destroyStreamFillSupporter(pInfo->pFillSup);

×

142

  blockDataDestroy(pInfo->pRes);

×

143

  pInfo->pRes = NULL;

×

144

  blockDataDestroy(pInfo->pSrcBlock);

×

145

  pInfo->pSrcBlock = NULL;

×

146

  blockDataDestroy(pInfo->pDelRes);

×

147

  pInfo->pDelRes = NULL;

×

148

  taosArrayDestroy(pInfo->matchInfo.pList);

×

149

  pInfo->matchInfo.pList = NULL;

×

150

  taosArrayDestroy(pInfo->pUpdated);

×

151

  clearGroupResInfo(&pInfo->groupResInfo);

×

152

  taosArrayDestroy(pInfo->pCloseTs);

×

154

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

×

155

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

×

158

  if (pInfo->pState != NULL) {

×

159

    taosMemoryFreeClear(pInfo->pState);

×

162

  taosMemoryFree(pInfo);

×

163

×

165

static void resetFillWindow(SResultRowData* pRowData) {

×

166

  pRowData->key = INT64_MIN;

×

167

  taosMemoryFreeClear(pRowData->pRowVal);

×

168

×

170

static void resetPrevAndNextWindow(SStreamFillSupporter* pFillSup) {

×

171

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

×

172

    resetFillWindow(&pFillSup->cur);

×

174

    pFillSup->cur.key = INT64_MIN;

×

175

    pFillSup->cur.pRowVal = NULL;

×

177

  resetFillWindow(&pFillSup->prev);

×

178

  resetFillWindow(&pFillSup->next);

×

179

  resetFillWindow(&pFillSup->nextNext);

×

180

×

182

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

×

183

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

×

184

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

×

185

  resetPrevAndNextWindow(pFillSup);

×

187

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

×

188

  void*   curVal = NULL;

×

189

  int32_t curVLen = 0;

×

190

  bool    hasCurKey = true;

×

191

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

×

192

  if (code == TSDB_CODE_SUCCESS) {

×

193

    pFillSup->cur.key = key.ts;

×

194

    pFillSup->cur.pRowVal = curVal;

×

196

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

×

197

    pFillSup->cur.key = ts;

×

198

    pFillSup->cur.pRowVal = NULL;

×

199

    hasCurKey = false;

×

202

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

×

203

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

×

204

  void*            preVal = NULL;

×

205

  int32_t          preVLen = 0;

×

206

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

×

208

  if (code == TSDB_CODE_SUCCESS) {

×

209

    pFillSup->prev.key = preKey.ts;

×

210

    pFillSup->prev.pRowVal = preVal;

×

212

    if (hasCurKey) {

×

213

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

×

216

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

×

218

    pAPI->stateStore.streamStateFreeCur(pCur);

×

219

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

×

222

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

×

223

  void*   nextVal = NULL;

×

224

  int32_t nextVLen = 0;

×

225

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

×

226

  if (code == TSDB_CODE_SUCCESS) {

×

227

    pFillSup->next.key = nextKey.ts;

×

228

    pFillSup->next.pRowVal = nextVal;

×

229

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

×

230

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

×

231

      SWinKey nextNextKey = {.groupId = groupId};

×

232

      void*   nextNextVal = NULL;

×

233

      int32_t nextNextVLen = 0;

×

234

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

×

235

      if (code == TSDB_CODE_SUCCESS) {

×

236

        pFillSup->nextNext.key = nextNextKey.ts;

×

237

        pFillSup->nextNext.pRowVal = nextNextVal;

×

241

  pAPI->stateStore.streamStateFreeCur(pCur);

×

242

×

244

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

×

245

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

×

246

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

×

247

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

×

249

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

×

250

  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);

×

251

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

×

252

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

×

253

    SResultCellData* pCell = getResultCell(pRowVal, i);

×

254

    if (!colDataIsNull_s(pColData, rowId)) {

×

255

      pCell->isNull = false;

×

256

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

×

257

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

×

258

      char* val = colDataGetData(pColData, rowId);

×

259

      if (IS_VAR_DATA_TYPE(pCell->type)) {

×

260

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

×

262

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

×

265

      pCell->isNull = true;

×

268

  pRowVal->key = ts;

×

269

×

271

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

×

272

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

×

273

    if (!pFillCol[i].notFillCol) {

×

274

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol + i);

×

275

      SResultCellData* pECell = getResultCell(pEndRow, slotId);

×

276

      SPoint*          pPoint = taosArrayGet(pEndPoins, slotId);

×

277

      pPoint->key = pEndRow->key;

×

278

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

×

281

×

283

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

×

284

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

×

285

  pFillInfo->start = ts;

×

286

×

288

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

×

289

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

×

290

  pFillInfo->end = ts;

×

291

×

293

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

×

294

  setFillInfoStart(start, pInterval, pFillInfo);

×

295

  pFillInfo->current = pFillInfo->start;

×

296

  setFillInfoEnd(end, pInterval, pFillInfo);

×

297

×

299

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

×

300

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

×

301

    pFillInfo->needFill = false;

×

302

    return;

×

305

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

×

306

                                pFillSup->interval.precision, NULL);

×

308

  pFillInfo->needFill = true;

×

309

  pFillInfo->start = realStart;

×

310

  pFillInfo->current = pFillInfo->start;

×

311

  pFillInfo->end = end;

×

312

  pFillInfo->pos = FILL_POS_INVALID;

×

313

  switch (pFillInfo->type) {

×

314

    case TSDB_FILL_NULL:

×

318

      break;

×

319

    case TSDB_FILL_PREV:

×

320

      pFillInfo->pResRow = &pFillSup->prev;

×

321

      break;

×

322

    case TSDB_FILL_NEXT:

×

323

      pFillInfo->pResRow = &pFillSup->next;

×

324

      break;

×

325

    case TSDB_FILL_LINEAR: {

×

326

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

×

327

      pFillInfo->pLinearInfo->hasNext = false;

×

328

      pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

329

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

×

331

      pFillInfo->pResRow = &pFillSup->prev;

×

332

      pFillInfo->pLinearInfo->winIndex = 0;

×

333

    } break;

×

334

    default:

×

335

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

×

336

      break;

×

340

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

×

341

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

×

342

    SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

343

    int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

344

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

×

345

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

×

346

    pCell->isNull = pCurCell->isNull;

×

347

    if (!pCurCell->isNull) {

×

348

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

×

351

×

353

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

×

355

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

×

356

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

×

357

    pFillInfo->needFill = false;

×

358

    pFillInfo->pos = FILL_POS_START;

×

359

    return;

×

361

  TSKEY prevWKey = INT64_MIN;

×

362

  TSKEY nextWKey = INT64_MIN;

×

363

  if (hasPrevWindow(pFillSup)) {

×

364

    prevWKey = pFillSup->prev.key;

×

366

  if (hasNextWindow(pFillSup)) {

×

367

    nextWKey = pFillSup->next.key;

×

370

  pFillInfo->needFill = true;

×

371

  pFillInfo->pos = FILL_POS_INVALID;

×

372

  switch (pFillInfo->type) {

×

373

    case TSDB_FILL_NULL:

×

377

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

×

378

        resetFillWindow(&pFillSup->prev);

×

380

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

×

381

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

×

382

          pFillInfo->preRowKey = INT64_MIN;

×

383

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

×

384

          pFillInfo->pos = FILL_POS_END;

×

386

          pFillInfo->needFill = false;

×

387

          pFillInfo->pos = FILL_POS_START;

×

389

      } else if (hasPrevWindow(pFillSup)) {

×

390

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

×

391

        pFillInfo->pos = FILL_POS_END;

×

393

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

×

394

        pFillInfo->pos = FILL_POS_START;

×

396

      copyNotFillExpData(pFillSup, pFillInfo);

×

397

    } break;

×

398

    case TSDB_FILL_PREV: {

×

399

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

×

400

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

×

401

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

×

402

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

×

403

        pFillInfo->pos = FILL_POS_START;

×

404

        resetFillWindow(&pFillSup->prev);

×

405

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

×

406

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

×

407

      } else if (hasPrevWindow(pFillSup)) {

×

408

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

×

409

        pFillInfo->pos = FILL_POS_END;

×

410

        pFillInfo->preRowKey = INT64_MIN;

×

412

      pFillInfo->pResRow = &pFillSup->prev;

×

413

    } break;

×

414

    case TSDB_FILL_NEXT: {

×

415

      if (hasPrevWindow(pFillSup)) {

×

416

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

×

417

        pFillInfo->pos = FILL_POS_END;

×

418

        resetFillWindow(&pFillSup->next);

×

419

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

×

420

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

×

421

        pFillInfo->preRowKey = INT64_MIN;

×

423

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

×

424

        pFillInfo->pos = FILL_POS_START;

×

426

      pFillInfo->pResRow = &pFillSup->next;

×

427

    } break;

×

428

    case TSDB_FILL_LINEAR: {

×

429

      pFillInfo->pLinearInfo->winIndex = 0;

×

430

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

×

431

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

×

432

        pFillInfo->pos = FILL_POS_MID;

×

433

        pFillInfo->pLinearInfo->nextEnd = nextWKey;

×

434

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

×

436

        pFillInfo->pResRow = &pFillSup->prev;

×

438

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

×

440

        pFillInfo->pLinearInfo->hasNext = true;

×

441

      } else if (hasPrevWindow(pFillSup)) {

×

442

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

×

443

        pFillInfo->pos = FILL_POS_END;

×

444

        pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

445

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

×

447

        pFillInfo->pResRow = &pFillSup->prev;

×

448

        pFillInfo->pLinearInfo->hasNext = false;

×

450

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

×

451

        pFillInfo->pos = FILL_POS_START;

×

452

        pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

453

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

×

455

        pFillInfo->pResRow = &pFillSup->cur;

×

456

        pFillInfo->pLinearInfo->hasNext = false;

×

458

    } break;

×

459

    default:

×

460

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

×

461

      break;

×

465

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

×

466

  int32_t code = TSDB_CODE_SUCCESS;

×

467

  int32_t lino = 0;

×

468

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

×

469

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

×

470

    (*pRes) = false;

×

471

    goto _end;

×

473

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

×

474

  QUERY_CHECK_CODE(code, lino, _end);

×

475

  (*pRes) = true;

×

477

_end:

×

478

  if (code != TSDB_CODE_SUCCESS) {

×

479

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

×

481

  return code;

×

484

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

×

486

  int32_t code = TSDB_CODE_SUCCESS;

×

487

  int32_t lino = 0;

×

488

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

×

489

    (*pRes) = false;

×

490

    goto _end;

×

492

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

×

493

  bool     ckRes = true;

×

494

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

×

495

  QUERY_CHECK_CODE(code, lino, _end);

×

497

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

×

498

    (*pRes) = true;

×

499

    goto _end;

×

501

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

×

502

    SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

503

    int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

504

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

×

505

    SFillInfo        tmpInfo = {

×

510

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

×

511

    if (!filled) {

×

512

      SResultCellData* pCell = getResultCell(pResRow, slotId);

×

513

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

×

514

      QUERY_CHECK_CODE(code, lino, _end);

×

517

  pBlock->info.rows++;

×

518

  (*pRes) = true;

×

520

_end:

×

521

  if (code != TSDB_CODE_SUCCESS) {

×

522

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

×

524

  return code;

×

527

bool hasRemainCalc(SStreamFillInfo* pFillInfo) {

×

528

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

×

529

    return true;

×

531

  return false;

×

534

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

×

535

  int32_t code = TSDB_CODE_SUCCESS;

×

536

  int32_t lino = 0;

×

537

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

×

538

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

×

539

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

×

540

      bool res = true;

×

541

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

×

542

      QUERY_CHECK_CODE(code, lino, _end);

×

544

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

×

545

                                     pFillSup->interval.precision, NULL);

×

548

_end:

×

549

  if (code != TSDB_CODE_SUCCESS) {

×

550

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

×

552

×

554

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

×

555

  int32_t code = TSDB_CODE_SUCCESS;

×

556

  int32_t lino = 0;

×

557

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

×

558

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

×

559

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

×

560

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

×

561

    bool        ckRes = true;

×

562

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

×

563

    QUERY_CHECK_CODE(code, lino, _end);

×

565

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

×

566

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

×

567

                                       pFillSup->interval.precision, NULL);

×

568

      pFillInfo->pLinearInfo->winIndex++;

×

569

      continue;

×

571

    pFillInfo->pLinearInfo->winIndex++;

×

572

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

×

573

      SFillColInfo* pFillCol = pFillSup->pAllColInfo + i;

×

574

      SFillInfo     tmp = {

×

575

              .currentKey = pFillInfo->current,

×

580

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

581

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

×

582

      int16_t          type = pColData->info.type;

×

583

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

×

584

      int32_t          index = pBlock->info.rows;

×

585

      if (pFillCol->notFillCol) {

×

586

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

×

587

        if (!filled) {

×

588

          code = setRowCell(pColData, index, pCell);

×

589

          QUERY_CHECK_CODE(code, lino, _end);

×

592

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

×

593

          colDataSetNULL(pColData, index);

×

594

          continue;

×

596

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

×

597

        double  vCell = 0;

×

598

        SPoint  start = {0};

×

599

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

×

600

        start.val = pCell->pData;

×

602

        SPoint cur = {0};

×

603

        cur.key = pFillInfo->current;

×

604

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

×

605

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

×

606

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

×

607

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

×

608

        QUERY_CHECK_CODE(code, lino, _end);

×

609

        destroySPoint(&cur);

×

612

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

×

613

                                     pFillSup->interval.precision, NULL);

×

614

    pBlock->info.rows++;

×

617

_end:

×

618

  if (code != TSDB_CODE_SUCCESS) {

×

619

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

×

621

×

623

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

×

624

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

×

626

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

×

627

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

×

628

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

×

629

  if (code != TSDB_CODE_SUCCESS) {

×

630

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

×

632

×

634

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

×

635

  int32_t code = TSDB_CODE_SUCCESS;

×

636

  int32_t lino = 0;

×

637

  bool    res = false;

×

638

  if (pFillInfo->needFill == false) {

×

639

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

×

640

    QUERY_CHECK_CODE(code, lino, _end);

×

641

    return;

×

644

  if (pFillInfo->pos == FILL_POS_START) {

×

645

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

×

646

    QUERY_CHECK_CODE(code, lino, _end);

×

647

    if (res) {

×

648

      pFillInfo->pos = FILL_POS_INVALID;

×

651

  if (pFillInfo->type != TSDB_FILL_LINEAR) {

×

652

    doStreamFillNormal(pFillSup, pFillInfo, pRes);

×

654

    doStreamFillLinear(pFillSup, pFillInfo, pRes);

×

656

    if (pFillInfo->pos == FILL_POS_MID) {

×

657

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

×

658

      QUERY_CHECK_CODE(code, lino, _end);

×

659

      if (res) {

×

660

        pFillInfo->pos = FILL_POS_INVALID;

×

664

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

×

665

      pFillInfo->pLinearInfo->hasNext = false;

×

666

      pFillInfo->pLinearInfo->winIndex = 0;

×

667

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

×

668

      pFillInfo->pResRow = &pFillSup->cur;

×

669

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

×

670

      doStreamFillLinear(pFillSup, pFillInfo, pRes);

×

673

  if (pFillInfo->pos == FILL_POS_END) {

×

674

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

×

675

    QUERY_CHECK_CODE(code, lino, _end);

×

676

    if (res) {

×

677

      pFillInfo->pos = FILL_POS_INVALID;

×

681

_end:

×

682

  if (code != TSDB_CODE_SUCCESS) {

×

683

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

×

687

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

×

689

  int32_t code = TSDB_CODE_SUCCESS;

×

690

  int32_t lino = 0;

×

691

  TSKEY ts = tsCol[rowId];

×

692

  pFillInfo->nextRowKey = ts;

×

693

  SResultRowData tmpNextRow = {.key = ts};

×

694

  tmpNextRow.pRowVal = taosMemoryCalloc(1, rowSize);

×

695

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

×

696

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

×

697

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

×

698

  taosMemoryFreeClear(tmpNextRow.pRowVal);

×

700

_end:

×

701

  if (code != TSDB_CODE_SUCCESS) {

×

702

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

×

704

  return code;

×

707

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

×

709

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

×

710

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

×

711

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

×

712

    resetFillWindow(&pFillSup->prev);

×

714

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

×

715

  doStreamFillRange(pFillInfo, pFillSup, pRes);

×

716

×

718

static void doStreamFillImpl(SOperatorInfo* pOperator) {

×

719

  int32_t                  code = TSDB_CODE_SUCCESS;

×

720

  int32_t                  lino = 0;

×

721

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

722

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

723

  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;

×

724

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

725

  SSDataBlock*             pBlock = pInfo->pSrcBlock;

×

726

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

×

727

  SSDataBlock*             pRes = pInfo->pRes;

×

728

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

×

729

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

×

730

  pRes->info.id.groupId = groupId;

×

731

  pInfo->srcRowIndex++;

×

733

  if (pInfo->srcRowIndex == 0) {

×

734

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

×

735

    QUERY_CHECK_CODE(code, lino, _end);

×

736

    pInfo->srcRowIndex++;

×

739

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

×

740

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

×

741

    QUERY_CHECK_CODE(code, lino, _end);

×

742

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

×

743

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

×

744

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

×

745

      QUERY_CHECK_CODE(code, lino, _end);

×

746

      return;

×

748

    pInfo->srcRowIndex++;

×

750

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

×

751

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

×

752

  QUERY_CHECK_CODE(code, lino, _end);

×

753

  blockDataCleanup(pInfo->pSrcBlock);

×

755

_end:

×

756

  if (code != TSDB_CODE_SUCCESS) {

×

757

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

×

761

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

×

762

  int32_t          code = TSDB_CODE_SUCCESS;

×

763

  int32_t          lino = 0;

×

764

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

×

765

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

×

766

  SExecTaskInfo*   pTaskInfo = pOp->pTaskInfo;

×

767

  SSDataBlock*     pBlock = delRes;

×

768

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

×

769

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

×

770

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

×

771

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

×

772

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

×

773

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

×

774

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

×

775

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

×

776

  QUERY_CHECK_CODE(code, lino, _end);

×

778

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

×

779

  QUERY_CHECK_CODE(code, lino, _end);

×

781

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

×

782

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

×

783

  QUERY_CHECK_CODE(code, lino, _end);

×

785

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

×

786

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

×

788

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

×

790

  void*   tbname = NULL;

×

791

  int32_t winCode = TSDB_CODE_SUCCESS;

×

792

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

×

793

  QUERY_CHECK_CODE(code, lino, _end);

×

794

  if (winCode != TSDB_CODE_SUCCESS) {

×

795

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

×

798

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

×

799

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

×

800

    QUERY_CHECK_CODE(code, lino, _end);

×

801

    pAPI->stateStore.streamStateFreeVal(tbname);

×

804

  pBlock->info.rows++;

×

806

_end:

×

807

  if (code != TSDB_CODE_SUCCESS) {

×

808

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

×

810

  return code;

×

813

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

×

815

  int32_t                  code = TSDB_CODE_SUCCESS;

×

816

  int32_t                  lino = 0;

×

817

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

818

  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;

×

819

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

820

  if (hasPrevWindow(pFillSup)) {

×

821

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

×

822

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

×

823

    QUERY_CHECK_CODE(code, lino, _end);

×

824

  } else if (hasNextWindow(pFillSup)) {

×

825

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

×

826

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

×

827

    QUERY_CHECK_CODE(code, lino, _end);

×

829

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

×

830

    QUERY_CHECK_CODE(code, lino, _end);

×

833

_end:

×

834

  if (code != TSDB_CODE_SUCCESS) {

×

835

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

×

837

  return code;

×

840

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

×

841

  int32_t                  code = TSDB_CODE_SUCCESS;

×

842

  int32_t                  lino = 0;

×

843

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

×

844

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

845

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

846

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

×

847

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

×

848

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

×

849

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

×

850

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

×

851

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

×

852

    QUERY_CHECK_CODE(code, lino, _end);

×

854

    STimeRange tw = {

×

859

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

×

860

    if (!tmp) {

×

861

      code = terrno;

×

862

      QUERY_CHECK_CODE(code, lino, _end);

×

866

_end:

×

867

  if (code != TSDB_CODE_SUCCESS) {

×

868

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

×

870

  return code;

×

873

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

×

874

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

×

875

  void*   val = NULL;

×

876

  int32_t len = 0;

×

877

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

×

878

  if (code != TSDB_CODE_SUCCESS) {

×

879

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

×

881

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

×

882

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

×

883

    pAPI->stateStore.streamStateFreeCur(pCur);

×

884

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

×

887

  if (code == TSDB_CODE_SUCCESS) {

×

888

    resetFillWindow(pWinData);

×

889

    pWinData->key = key.ts;

×

890

    pWinData->pRowVal = val;

×

892

×

894

static void doDeleteFillFinalize(SOperatorInfo* pOperator) {

×

895

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

×

897

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

898

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

899

  int32_t                  size = taosArrayGetSize(pFillInfo->delRanges);

×

900

  while (pFillInfo->delIndex < size) {

×

901

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

×

902

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

×

903

      return;

×

905

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

×

906

    TSKEY realEnd = range->ekey + 1;

×

907

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

×

908

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

×

909

                         &pInfo->pFillSup->next);

×

911

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

×

912

    pFillInfo->delIndex++;

×

913

    if (pInfo->pFillInfo->needFill) {

×

914

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

×

915

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

×

920

static int32_t doDeleteFillResult(SOperatorInfo* pOperator) {

×

921

  int32_t                  code = TSDB_CODE_SUCCESS;

×

922

  int32_t                  lino = 0;

×

923

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

×

924

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

925

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

926

  SSDataBlock*             pBlock = pInfo->pSrcDelBlock;

×

927

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

929

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

×

930

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

×

931

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

×

932

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

×

933

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

×

934

    TSKEY            ts = tsStarts[pInfo->srcDelRowIndex];

×

935

    TSKEY            endTs = ts;

×

936

    uint64_t         groupId = groupIds[pInfo->srcDelRowIndex];

×

937

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

×

938

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

×

940

    if (!pCur) {

×

941

      pInfo->srcDelRowIndex++;

×

942

      continue;

×

945

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

×

946

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

×

947

      TSKEY    delTs = tsStarts[pInfo->srcDelRowIndex];

×

948

      uint64_t delGroupId = groupIds[pInfo->srcDelRowIndex];

×

949

      int32_t  winCode = TSDB_CODE_SUCCESS;

×

950

      if (groupId != delGroupId) {

×

951

        break;

×

953

      if (delTs > nextKey.ts) {

×

954

        break;

×

957

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

×

958

      if (delTs == nextKey.ts) {

×

959

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

×

960

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

×

962

        if (delTs != ts) {

×

963

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

×

964

          pAPI->stateStore.streamStateFreeCur(pCur);

×

965

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

×

967

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

×

968

        if (winCode != TSDB_CODE_SUCCESS) {

×

969

          break;

×

972

      pInfo->srcDelRowIndex++;

×

975

    pAPI->stateStore.streamStateFreeCur(pCur);

×

976

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

×

977

    QUERY_CHECK_CODE(code, lino, _end);

×

980

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

×

982

_end:

×

983

  if (code != TSDB_CODE_SUCCESS) {

×

984

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

×

986

  return code;

×

989

void resetStreamFillSup(SStreamFillSupporter* pFillSup) {

×

990

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

991

  SSHashObj* pNewMap = tSimpleHashInit(16, hashFn);

×

992

  if (pNewMap != NULL) {

×

993

    tSimpleHashCleanup(pFillSup->pResMap);

×

994

    pFillSup->pResMap = pNewMap;

×

996

    tSimpleHashClear(pFillSup->pResMap);

×

998

  pFillSup->hasDelete = false;

×

999

×

1000

void resetStreamFillInfo(SStreamFillOperatorInfo* pInfo) {

×

1001

  resetStreamFillSup(pInfo->pFillSup);

×

1002

  taosArrayClear(pInfo->pFillInfo->delRanges);

×

1003

  pInfo->pFillInfo->delIndex = 0;

×

1004

×

1006

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

×

1008

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1009

  int32_t                  lino = 0;

×

1010

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1011

  SExprSupp*               pSup = &pOperator->exprSupp;

×

1012

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1014

  blockDataCleanup(pDstBlock);

×

1015

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

×

1016

  QUERY_CHECK_CODE(code, lino, _end);

×

1018

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

×

1019

  QUERY_CHECK_CODE(code, lino, _end);

×

1020

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

×

1021

  QUERY_CHECK_CODE(code, lino, _end);

×

1023

  pDstBlock->info.rows = 0;

×

1024

  pSup = &pInfo->pFillSup->notFillExprSup;

×

1025

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

×

1026

  QUERY_CHECK_CODE(code, lino, _end);

×

1027

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

×

1028

  QUERY_CHECK_CODE(code, lino, _end);

×

1030

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

×

1032

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

×

1034

_end:

×

1035

  if (code != TSDB_CODE_SUCCESS) {

×

1036

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

×

1038

  return code;

×

1041

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

×

1042

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1043

  int32_t                  lino = 0;

×

1044

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1045

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1047

  if (pOperator->status == OP_EXEC_DONE) {

×

1048

    (*ppRes) = NULL;

×

1049

    return code;

×

1051

  blockDataCleanup(pInfo->pRes);

×

1052

  if (hasRemainCalc(pInfo->pFillInfo) ||

×

1053

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

×

1054

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

×

1055

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

×

1056

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

×

1057

      (*ppRes) = pInfo->pRes;

×

1058

      return code;

×

1061

  if (pOperator->status == OP_RES_TO_RETURN) {

×

1062

    doDeleteFillFinalize(pOperator);

×

1063

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

×

1064

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

×

1065

      (*ppRes) = pInfo->pRes;

×

1066

      return code;

×

1068

    setOperatorCompleted(pOperator);

×

1069

    resetStreamFillInfo(pInfo);

×

1070

    (*ppRes) = NULL;

×

1071

    return code;

×

1074

  SSDataBlock*   fillResult = NULL;

×

1075

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

×

1077

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

×

1079

      SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

1080

      if (pBlock == NULL) {

×

1081

        pOperator->status = OP_RES_TO_RETURN;

×

1082

        pInfo->pFillInfo->preRowKey = INT64_MIN;

×

1083

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

×

1084

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

×

1085

          (*ppRes) = pInfo->pRes;

×

1086

          return code;

×

1088

        break;

×

1090

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

×

1092

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

×

1093

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

×

1094

        pInfo->pFillInfo->preRowKey = INT64_MIN;

×

1097

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

×

1098

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

×

1099

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

×

1102

      switch (pBlock->info.type) {

×

1103

        case STREAM_RETRIEVE:

×

1104

          (*ppRes) = pBlock;

×

1105

          return code;

×

1106

        case STREAM_DELETE_RESULT: {

×

1107

          pInfo->pSrcDelBlock = pBlock;

×

1108

          pInfo->srcDelRowIndex = 0;

×

1109

          blockDataCleanup(pInfo->pDelRes);

×

1110

          pInfo->pFillSup->hasDelete = true;

×

1111

          code = doDeleteFillResult(pOperator);

×

1112

          QUERY_CHECK_CODE(code, lino, _end);

×

1114

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

×

1115

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

×

1116

            (*ppRes) = pInfo->pDelRes;

×

1117

            return code;

×

1119

          continue;

×

1121

        case STREAM_NORMAL:

×

1124

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

×

1125

          QUERY_CHECK_CODE(code, lino, _end);

×

1127

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

×

1128

          pInfo->srcRowIndex = -1;

×

1129

        } break;

×

1130

        case STREAM_CHECKPOINT:

×

1132

          (*ppRes) = pBlock;

×

1133

          return code;

×

1135

        default:

×

1136

          return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1140

    doStreamFillImpl(pOperator);

×

1141

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

×

1142

    QUERY_CHECK_CODE(code, lino, _end);

×

1144

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

×

1145

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

×

1146

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

×

1147

      break;

×

1150

  if (pOperator->status == OP_RES_TO_RETURN) {

×

1151

    doDeleteFillFinalize(pOperator);

×

1154

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

×

1155

    setOperatorCompleted(pOperator);

×

1156

    resetStreamFillInfo(pInfo);

×

1157

    (*ppRes) = NULL;

×

1158

    return code;

×

1161

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

×

1162

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

×

1163

  (*ppRes) = pInfo->pRes;

×

1164

  return code;

×

1166

_end:

×

1167

  if (code != TSDB_CODE_SUCCESS) {

×

1168

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

×

1169

    pTaskInfo->code = code;

×

1170

    T_LONG_JMP(pTaskInfo->env, code);

×

1172

  setOperatorCompleted(pOperator);

×

1173

  resetStreamFillInfo(pInfo);

×

1174

  (*ppRes) = NULL;

×

1175

  return code;

×

1178

static void resetForceFillWindow(SResultRowData* pRowData) {

×

1179

  pRowData->key = INT64_MIN;

×

1180

  pRowData->pRowVal = NULL;

×

1181

×

1183

void doBuildForceFillResultImpl(SOperatorInfo* pOperator, SStreamFillSupporter* pFillSup,

×

1185

  int32_t code = TSDB_CODE_SUCCESS;

×

1186

  int32_t lino = 0;

×

1188

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1189

  bool                     res = false;

×

1190

  int32_t                  numOfRows = getNumOfTotalRes(pGroupResInfo);

×

1191

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

×

1192

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

×

1193

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

×

1194

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

×

1195

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

×

1196

      break;

×

1199

    SRowBuffPos* pValPos = NULL;

×

1200

    int32_t      len = 0;

×

1201

    int32_t      winCode = TSDB_CODE_SUCCESS;

×

1202

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

×

1203

    QUERY_CHECK_CODE(code, lino, _end);

×

1204

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

×

1205

    if (winCode == TSDB_CODE_SUCCESS) {

×

1206

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

×

1207

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

×

1208

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

×

1209

      QUERY_CHECK_CODE(code, lino, _end);

×

1210

      resetForceFillWindow(&pFillSup->cur);

×

1211

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

×

1213

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

×

1214

      SRowBuffPos* prePos = NULL;

×

1215

      int32_t      preVLen = 0;

×

1216

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

×

1218

      QUERY_CHECK_CODE(code, lino, _end);

×

1219

      if (winCode == TSDB_CODE_SUCCESS) {

×

1220

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

×

1221

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

×

1222

        if (pFillInfo->type == TSDB_FILL_PREV) {

×

1223

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

×

1224

          QUERY_CHECK_CODE(code, lino, _end);

×

1226

          copyNotFillExpData(pFillSup, pFillInfo);

×

1227

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

×

1228

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

×

1229

          QUERY_CHECK_CODE(code, lino, _end);

×

1231

        resetForceFillWindow(&pFillSup->cur);

×

1233

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

×

1237

_end:

×

1238

  if (code != TSDB_CODE_SUCCESS) {

×

1239

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

×

1241

×

1243

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

×

1245

  blockDataCleanup(pBlock);

×

1246

  if (!hasRemainResults(pGroupResInfo)) {

×

1247

    return;

×

1251

  pBlock->info.id.groupId = 0;

×

1252

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

×

1255

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

×

1256

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1257

  int32_t                  lino = 0;

×

1258

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1259

  uint16_t                 opType = pOperator->operatorType;

×

1260

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1262

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

×

1263

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

×

1264

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

×

1265

    (*ppRes) = pInfo->pRes;

×

1266

    goto _end;

×

1269

  (*ppRes) = NULL;

×

1271

_end:

×

1272

  if (code != TSDB_CODE_SUCCESS) {

×

1273

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

×

1275

  return code;

×

1278

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

×

1279

  int32_t code = TSDB_CODE_SUCCESS;

×

1280

  int32_t lino = 0;

×

1282

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

×

1283

  int32_t      curVLen = 0;

×

1284

  SRowBuffPos* pStatePos = NULL;

×

1285

  int32_t      winCode = TSDB_CODE_SUCCESS;

×

1286

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

×

1288

  QUERY_CHECK_CODE(code, lino, _end);

×

1289

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

×

1290

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

×

1292

_end:

×

1293

  if (code != TSDB_CODE_SUCCESS) {

×

1294

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

×

1296

×

1298

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

×

1300

  int32_t code = TSDB_CODE_SUCCESS;

×

1301

  int32_t lino = 0;

×

1302

  TSKEY ts = tsCol[rowId];

×

1303

  pFillInfo->nextRowKey = ts;

×

1304

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

×

1305

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

×

1307

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

×

1308

  keepResultInStateBuf(pInfo, groupId, &tmpNextRow);

×

1310

_end:

×

1311

  if (code != TSDB_CODE_SUCCESS) {

×

1312

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

×

1314

  return code;

×

1318

static int32_t doStreamForceFillImpl(SOperatorInfo* pOperator) {

×

1319

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1320

  int32_t                  lino = 0;

×

1321

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1322

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1323

  SStreamFillSupporter*    pFillSup = pInfo->pFillSup;

×

1324

  SStreamFillInfo*         pFillInfo = pInfo->pFillInfo;

×

1325

  SSDataBlock*             pBlock = pInfo->pSrcBlock;

×

1326

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

×

1327

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

×

1328

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

×

1329

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

×

1330

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

×

1331

    QUERY_CHECK_CODE(code, lino, _end);

×

1333

    int32_t size =  taosArrayGetSize(pInfo->pCloseTs);

×

1334

    if (size > 0) {

×

1335

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

×

1336

      TSKEY  resTs = tsCol[i];

×

1337

      while (resTs < (*pTs)) {

×

1338

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

×

1339

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

×

1340

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

×

1342

        if (IS_FILL_CONST_VALUE(pFillSup->type)) {

×

1345

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

×

1346

                            pFillSup->interval.precision, NULL);

×

1350

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

×

1351

  QUERY_CHECK_CODE(code, lino, _end);

×

1353

_end:

×

1354

  if (code != TSDB_CODE_SUCCESS) {

×

1355

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

×

1357

  return code;

×

1360

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

×

1361

  int32_t          code = TSDB_CODE_SUCCESS;

×

1362

  int32_t          lino = 0;

×

1363

  int64_t          groupId = 0;

×

1364

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

×

1365

  while (1) {

×

1366

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

×

1367

    if (winCode != TSDB_CODE_SUCCESS) {

×

1368

      break;

×

1370

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

×

1371

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

×

1372

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

×

1374

    pStateStore->streamStateGroupCurNext(pCur);

×

1376

  pStateStore->streamStateFreeCur(pCur);

×

1377

  pCur = NULL;

×

1379

_end:

×

1380

  if (code != TSDB_CODE_SUCCESS) {

×

1381

    pStateStore->streamStateFreeCur(pCur);

×

1382

    pCur = NULL;

×

1383

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

×

1385

  return code;

×

1388

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

×

1389

  taosArraySort(pTsArrray, fn);

×

1390

  taosArrayRemoveDuplicate(pTsArrray, fn, NULL);

×

1391

×

1394

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

×

1395

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1396

  int32_t                  lino = 0;

×

1397

  SStreamFillOperatorInfo* pInfo = pOperator->info;

×

1398

  SExecTaskInfo*           pTaskInfo = pOperator->pTaskInfo;

×

1400

  if (pOperator->status == OP_EXEC_DONE) {

×

1401

    (*ppRes) = NULL;

×

1402

    return code;

×

1405

  if (pOperator->status == OP_RES_TO_RETURN) {

×

1406

    SSDataBlock* resBlock = NULL;

×

1407

    code = buildForceFillResult(pOperator, &resBlock);

×

1408

    QUERY_CHECK_CODE(code, lino, _end);

×

1410

    if (resBlock != NULL) {

×

1411

      (*ppRes) = resBlock;

×

1412

      goto _end;

×

1415

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

×

1416

    resetStreamFillInfo(pInfo);

×

1417

    setStreamOperatorCompleted(pOperator);

×

1418

    (*ppRes) = NULL;

×

1419

    goto _end;

×

1422

  SSDataBlock*   fillResult = NULL;

×

1423

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

×

1424

  while (1) {

×

1425

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

1426

    if (pBlock == NULL) {

×

1427

      pOperator->status = OP_RES_TO_RETURN;

×

1428

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

×

1429

      break;

×

1431

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

×

1432

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

×

1434

    switch (pBlock->info.type) {

×

1435

      case STREAM_NORMAL:

×

1437

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

×

1438

        QUERY_CHECK_CODE(code, lino, _end);

×

1440

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

×

1441

        pInfo->srcRowIndex = -1;

×

1442

      } break;

×

1443

      case STREAM_CHECKPOINT: {

×

1444

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

×

1445

        (*ppRes) = pBlock;

×

1446

        goto _end;

×

1448

      case STREAM_CREATE_CHILD_TABLE: {

×

1449

        (*ppRes) = pBlock;

×

1450

        goto _end;

×

1452

      case STREAM_GET_RESULT: {

×

1453

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

×

1454

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

×

1455

        continue;

×

1457

      default:

×

1458

        code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1459

        QUERY_CHECK_CODE(code, lino, _end);

×

1462

    code = doStreamForceFillImpl(pOperator);

×

1463

    QUERY_CHECK_CODE(code, lino, _end);

×

1466

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

×

1467

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

×

1468

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

×

1469

    QUERY_CHECK_CODE(code, lino, _end);

×

1471

  taosArrayClear(pInfo->pCloseTs);

×

1472

  removeDuplicateResult(pInfo->pUpdated, winKeyCmprImpl);

×

1474

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

×

1475

  pInfo->groupResInfo.freeItem = false;

×

1477

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

×

1478

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

×

1480

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

×

1481

  QUERY_CHECK_CODE(code, lino, _end);

×

1483

  code = buildForceFillResult(pOperator, ppRes);

×

1484

  QUERY_CHECK_CODE(code, lino, _end);

×

1486

  if ((*ppRes) == NULL) {

×

1487

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

×

1488

    resetStreamFillInfo(pInfo);

×

1489

    setStreamOperatorCompleted(pOperator);

×

1492

_end:

×

1493

  if (code != TSDB_CODE_SUCCESS) {

×

1494

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

×

1495

    pTaskInfo->code = code;

×

1497

  return code;

×

1500

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

×

1501

  int32_t numOfCols = taosArrayGetSize(pInputRes->pDataBlock);

×

1502

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

×

1503

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

×

1504

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

×

1505

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

×

1507

  pFillSup->next.key = INT64_MIN;

×

1508

  pFillSup->nextNext.key = INT64_MIN;

×

1509

  pFillSup->prev.key = INT64_MIN;

×

1510

  pFillSup->cur.key = INT64_MIN;

×

1511

  pFillSup->next.pRowVal = NULL;

×

1512

  pFillSup->nextNext.pRowVal = NULL;

×

1513

  pFillSup->prev.pRowVal = NULL;

×

1514

  pFillSup->cur.pRowVal = NULL;

×

1516

  return TSDB_CODE_SUCCESS;

×

1519

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

×

1521

  int32_t               code = TSDB_CODE_SUCCESS;

×

1522

  int32_t               lino = 0;

×

1523

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

×

1524

  if (!pFillSup) {

×

1525

    code = terrno;

×

1526

    QUERY_CHECK_CODE(code, lino, _end);

×

1528

  pFillSup->numOfFillCols = numOfFillCols;

×

1529

  int32_t    numOfNotFillCols = 0;

×

1530

  SExprInfo* noFillExprInfo = NULL;

×

1532

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

×

1533

  QUERY_CHECK_CODE(code, lino, _end);

×

1535

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

×

1536

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

×

1537

  if (pFillSup->pAllColInfo == NULL) {

×

1538

    code = terrno;

×

1539

    lino = __LINE__;

×

1540

    destroyExprInfo(noFillExprInfo, numOfNotFillCols);

×

1541

    goto _end;

×

1544

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

×

1545

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

×

1546

  pFillSup->interval = *pInterval;

×

1547

  pFillSup->pAPI = pAPI;

×

1549

  code = initResultBuf(pInputRes, pFillSup);

×

1550

  QUERY_CHECK_CODE(code, lino, _end);

×

1552

  SExprInfo* noFillExpr = NULL;

×

1553

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

×

1554

  QUERY_CHECK_CODE(code, lino, _end);

×

1556

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

×

1557

  QUERY_CHECK_CODE(code, lino, _end);

×

1559

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

1560

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

×

1561

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

×

1562

  pFillSup->hasDelete = false;

×

1564

_end:

×

1565

  if (code != TSDB_CODE_SUCCESS) {

×

1566

    destroyStreamFillSupporter(pFillSup);

×

1567

    pFillSup = NULL;

×

1568

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

×

1570

  return pFillSup;

×

1573

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

×

1574

  int32_t          code = TSDB_CODE_SUCCESS;

×

1575

  int32_t          lino = 0;

×

1576

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

×

1577

  if (!pFillInfo) {

×

1578

    code = terrno;

×

1579

    QUERY_CHECK_CODE(code, lino, _end);

×

1582

  pFillInfo->start = INT64_MIN;

×

1583

  pFillInfo->current = INT64_MIN;

×

1584

  pFillInfo->end = INT64_MIN;

×

1585

  pFillInfo->preRowKey = INT64_MIN;

×

1586

  pFillInfo->needFill = false;

×

1587

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

×

1588

  if (!pFillInfo) {

×

1589

    code = terrno;

×

1590

    QUERY_CHECK_CODE(code, lino, _end);

×

1593

  pFillInfo->pLinearInfo->hasNext = false;

×

1594

  pFillInfo->pLinearInfo->nextEnd = INT64_MIN;

×

1595

  pFillInfo->pLinearInfo->pEndPoints = NULL;

×

1596

  pFillInfo->pLinearInfo->pNextEndPoints = NULL;

×

1597

  if (pFillSup->type == TSDB_FILL_LINEAR) {

×

1598

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

×

1599

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

×

1600

      code = terrno;

×

1601

      QUERY_CHECK_CODE(code, lino, _end);

×

1604

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

×

1605

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

×

1606

      code = terrno;

×

1607

      QUERY_CHECK_CODE(code, lino, _end);

×

1610

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

×

1611

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

×

1612

      if (pColData == NULL) {

×

1613

        SPoint dummy = {0};

×

1614

        dummy.val = taosMemoryCalloc(1, 1);

×

1615

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

×

1616

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

×

1618

        dummy.val = taosMemoryCalloc(1, 1);

×

1619

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

×

1620

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

×

1622

        continue;

×

1624

      SPoint value = {0};

×

1625

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

×

1626

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

×

1628

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

×

1629

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

×

1631

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

×

1632

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

×

1634

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

×

1635

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

×

1638

  pFillInfo->pLinearInfo->winIndex = 0;

×

1640

  pFillInfo->pNonFillRow = NULL;

×

1641

  pFillInfo->pResRow = NULL;

×

1642

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

×

1643

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

×

1644

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

×

1645

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

×

1647

    pFillInfo->pResRow->key = INT64_MIN;

×

1648

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

×

1649

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

×

1651

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

×

1652

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

×

1653

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

×

1654

      if (pColData == NULL) {

×

1655

        pCell->bytes = 1;

×

1656

        pCell->type = 4;

×

1657

        continue;

×

1659

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

×

1660

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

×

1663

    int32_t numOfResCol = taosArrayGetSize(pRes->pDataBlock);

×

1664

    if (numOfResCol < pFillSup->numOfAllCols) {

×

1665

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

×

1666

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

×

1667

      pFillSup->pOffsetInfo = pTmpBuf;

×

1669

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

×

1670

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

×

1671

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

×

1672

        pFillSup->pOffsetInfo[i] = preLength;

×

1673

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

×

1674

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

×

1678

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

×

1679

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

×

1680

    pFillInfo->pNonFillRow->key = INT64_MIN;

×

1681

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

×

1682

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

×

1685

  pFillInfo->type = pFillSup->type;

×

1686

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

×

1687

  if (!pFillInfo->delRanges) {

×

1688

    code = terrno;

×

1689

    QUERY_CHECK_CODE(code, lino, _end);

×

1692

  pFillInfo->delIndex = 0;

×

1693

  pFillInfo->curGroupId = 0;

×

1694

  pFillInfo->hasNext = false;

×

1695

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

×

1696

  return pFillInfo;

×

1698

_end:

×

1699

  if (code != TSDB_CODE_SUCCESS) {

×

1700

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

×

1702

  destroyStreamFillInfo(pFillInfo);

×

1703

  return NULL;

×

1706

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

×

1707

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

×

1708

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

×

1709

      SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

1710

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

1711

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

×

1712

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

×

1713

      if (pCell->type == TSDB_DATA_TYPE_FLOAT) {

×

1714

        float v = 0;

×

1715

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

×

1716

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

×

1717

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

×

1718

        double v = 0;

×

1719

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

×

1720

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

×

1721

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

×

1722

        int64_t v = 0;

×

1723

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

×

1724

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

×

1726

        pCell->isNull = true;

×

1729

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

×

1730

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

×

1731

      SFillColInfo*    pFillCol = pFillSup->pAllColInfo + i;

×

1732

      int32_t          slotId = GET_DEST_SLOT_ID(pFillCol);

×

1733

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

×

1734

      pCell->isNull = true;

×

1737

×

1739

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

×

1740

  int32_t code = TSDB_CODE_SUCCESS;

×

1741

  int32_t lino = 0;

×

1742

  if (IS_NORMAL_INTERVAL_OP(downstream)) {

×

1743

    SStreamIntervalOperatorInfo* pInfo = downstream->info;

×

1744

    *triggerType = pInfo->twAggSup.calTrigger;

×

1745

    *pInterval = pInfo->interval;

×

1746

  } else if (IS_CONTINUE_INTERVAL_OP(downstream)) {

×

1747

    SStreamIntervalSliceOperatorInfo* pInfo = downstream->info;

×

1748

    *triggerType = pInfo->twAggSup.calTrigger;

×

1749

    *pInterval = pInfo->interval;

×

1750

    pInfo->hasFill = true;

×

1752

    code = TSDB_CODE_STREAM_INTERNAL_ERROR;

×

1754

  QUERY_CHECK_CODE(code, lino, _end);

×

1756

_end:

×

1757

  if (code != TSDB_CODE_SUCCESS) {

×

1758

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

×

1760

  return code;

×

1763

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

×

1765

  int32_t code = TSDB_CODE_SUCCESS;

×

1766

  int32_t lino = 0;

×

1768

  pInfo->stateStore = *pStore;

×

1769

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

×

1770

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

×

1772

  *(pInfo->pState) = *pState;

×

1773

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

×

1774

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

×

1775

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

×

1776

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

×

1777

  QUERY_CHECK_CODE(code, lino, _end);

×

1779

_end:

×

1780

  if (code != TSDB_CODE_SUCCESS) {

×

1781

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

×

1783

  return code;

×

1786

int32_t createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode,

×

1788

  QRY_PARAM_CHECK(pOptrInfo);

×

1790

  int32_t                  code = TSDB_CODE_SUCCESS;

×

1791

  int32_t                  lino = 0;

×

1792

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

×

1793

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

×

1794

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

×

1795

    code = terrno;

×

1796

    QUERY_CHECK_CODE(code, lino, _error);

×

1799

  int32_t    numOfFillCols = 0;

×

1800

  SExprInfo* pFillExprInfo = NULL;

×

1802

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

×

1803

  QUERY_CHECK_CODE(code, lino, _error);

×

1805

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

×

1806

  QUERY_CHECK_CODE(code, lino, _error);

×

1808

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

×

1809

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

×

1811

  int8_t triggerType = 0;

×

1812

  SInterval interval = {0};

×

1813

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

×

1814

  QUERY_CHECK_CODE(code, lino, _error);

×

1816

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

×

1818

  if (!pInfo->pFillSup) {

×

1819

    code = TSDB_CODE_FAILED;

×

1820

    QUERY_CHECK_CODE(code, lino, _error);

×

1823

  initResultSizeInfo(&pOperator->resultInfo, 4096);

×

1824

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

×

1825

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

×

1827

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

×

1828

  QUERY_CHECK_CODE(code, lino, _error);

×

1830

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

×

1831

  QUERY_CHECK_CODE(code, lino, _error);

×

1833

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

×

1834

  if (!pInfo->pFillInfo) {

×

1835

    goto _error;

×

1838

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

×

1840

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

×

1841

  QUERY_CHECK_CODE(code, lino, _error);

×

1843

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

×

1844

  QUERY_CHECK_CODE(code, lino, _error);

×

1846

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

×

1847

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

×

1849

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

×

1850

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

×

1852

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

×

1853

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

×

1855

  int32_t numOfOutputCols = 0;

×

1856

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

×

1858

  QUERY_CHECK_CODE(code, lino, _error);

×

1860

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

×

1861

  QUERY_CHECK_CODE(code, lino, _error);

×

1863

  pInfo->srcRowIndex = -1;

×

1864

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

×

1867

  if (triggerType == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {

×

1868

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

×

1869

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

×

1870

    QUERY_CHECK_CODE(code, lino, _error);

×

1871

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

×

1874

    pInfo->pState = NULL;

×

1875

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

×

1878

  setOperatorStreamStateFn(pOperator, streamOpReleaseState, streamOpReloadState);

×

1880

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

×

1881

  QUERY_CHECK_CODE(code, lino, _error);

×

1883

  *pOptrInfo = pOperator;

×

1884

  return TSDB_CODE_SUCCESS;

×

1886

_error:

×

1887

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

×

1889

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

×

1890

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);

×

1891

  pTaskInfo->code = code;

×

1892

  return code;

×

taosdata / TDengine / #3633

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

taosdata / TDengine / #3633

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