taosdata/TDengine | Build 4979 | source/libs/executor/src/eventwindowoperator.c | Coveralls

32

static int32_t resetEventWindowOperState(SOperatorInfo* pOper) {

1,494✔

33

  SEventWindowOperatorInfo* pEvent = pOper->info;

1,494✔

34

  SExecTaskInfo*           pTaskInfo = pOper->pTaskInfo;

1,494✔

35

  SEventWinodwPhysiNode* pPhynode = (SEventWinodwPhysiNode*)pOper->pPhyNode;

1,494✔

36

  pOper->status = OP_NOT_OPENED;

1,494✔

38

  resetBasicOperatorState(&pEvent->binfo);

1,494✔

39

  taosMemoryFreeClear(pEvent->pRow);

1,494✔

41

  pEvent->groupId = 0;

1,494✔

42

  pEvent->pPreDataBlock = NULL;

1,494✔

43

  pEvent->inWindow = false;

1,494✔

44

  pEvent->winSup.lastTs = INT64_MIN;

1,494✔

46

  colDataDestroy(&pEvent->twAggSup.timeWindowData);

1,494✔

47

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

1,494✔

48

  cleanupResultInfoInEventWindow(pOper, pEvent);

1,494✔

50

  if (code == 0) {

1,494✔

51

    code = resetAggSup(&pOper->exprSupp, &pEvent->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,

2,988✔

52

                       sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, pTaskInfo->streamInfo.pState,

1,494✔

55

  if (code == 0) {

1,494✔

56

    code = resetExprSupp(&pEvent->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,

1,494✔

59

  return code;

1,494✔

62

int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,

294,646✔

64

  QRY_PARAM_CHECK(pOptrInfo);

294,646✔

66

  int32_t                   code = TSDB_CODE_SUCCESS;

294,646✔

67

  int32_t                   lino = 0;

294,646✔

68

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

294,646✔

69

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

294,256✔

70

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

294,256✔

75

  pOperator->pPhyNode = physiNode;

294,256✔

76

  pOperator->exprSupp.hasWindowOrGroup = true;

294,256✔

77

  pOperator->exprSupp.hasWindow = true;

294,256✔

79

  SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;

294,256✔

81

  int32_t tsSlotId = ((SColumnNode*)pEventWindowNode->window.pTspk)->slotId;

294,256✔

82

  code = filterInitFromNode((SNode*)pEventWindowNode->pStartCond, &pInfo->pStartCondInfo, 0,

294,646✔

83

                            pTaskInfo->pStreamRuntimeInfo);

294,646✔

84

  QUERY_CHECK_CODE(code, lino, _error);

294,646✔

86

  code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0,

294,646✔

87

                            pTaskInfo->pStreamRuntimeInfo);

294,646✔

88

  QUERY_CHECK_CODE(code, lino, _error);

294,646✔

90

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

294,646✔

91

    int32_t    numOfScalarExpr = 0;

63,959✔

92

    SExprInfo* pScalarExprInfo = NULL;

63,959✔

94

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

63,959✔

95

    QUERY_CHECK_CODE(code, lino, _error);

63,959✔

96

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

63,959✔

97

    QUERY_CHECK_CODE(code, lino, _error);

63,959✔

100

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

294,646✔

101

                            pTaskInfo->pStreamRuntimeInfo);

294,646✔

102

  QUERY_CHECK_CODE(code, lino, _error);

294,646✔

104

  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;

294,646✔

106

  int32_t    num = 0;

294,646✔

107

  SExprInfo* pExprInfo = NULL;

294,646✔

108

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

294,646✔

109

  QUERY_CHECK_CODE(code, lino, _error);

294,646✔

111

  initResultSizeInfo(&pOperator->resultInfo, 4096);

294,646✔

113

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

589,292✔

114

                    pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);

294,646✔

115

  QUERY_CHECK_CODE(code, lino, _error);

294,646✔

117

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

294,646✔

118

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

294,646✔

119

  initBasicInfo(&pInfo->binfo, pResBlock);

294,646✔

121

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

294,646✔

122

  QUERY_CHECK_CODE(code, lino, _error);

294,646✔

124

  initResultRowInfo(&pInfo->binfo.resultRowInfo);

294,646✔

125

  pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;

294,646✔

126

  pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;

294,646✔

127

  pInfo->winSup.lastTs = INT64_MIN;

294,646✔

129

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

294,646✔

130

  QUERY_CHECK_CODE(code, lino, _error);

294,646✔

132

  pInfo->tsSlotId = tsSlotId;

294,646✔

133

  pInfo->pPreDataBlock = NULL;

294,646✔

134

  pInfo->pOperator = pOperator;

294,646✔

135

  pInfo->trueForInfo.trueForType = pEventWindowNode->trueForType;

294,646✔

136

  pInfo->trueForInfo.count = pEventWindowNode->trueForCount;

294,646✔

137

  pInfo->trueForInfo.duration = pEventWindowNode->trueForDuration;

294,646✔

139

  setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT, true, OP_NOT_OPENED, pInfo,

294,646✔

141

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, eventWindowAggregateNext, NULL, destroyEWindowOperatorInfo,

294,646✔

144

  setOperatorResetStateFn(pOperator, resetEventWindowOperState);

294,646✔

145

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

294,646✔

146

  if (code != TSDB_CODE_SUCCESS) {

294,646✔

150

  *pOptrInfo = pOperator;

294,646✔

151

  return TSDB_CODE_SUCCESS;

294,646✔

163

void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo) {

296,140✔

164

  if (pInfo == NULL || pInfo->pRow == NULL || pOperator == NULL) {

296,140✔

165

    return;

84,097✔

167

  SExprSupp*       pSup = &pOperator->exprSupp;

212,043✔

168

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

212,043✔

176

void destroyEWindowOperatorInfo(void* param) {

294,646✔

177

  SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;

294,646✔

178

  if (pInfo == NULL) {

294,646✔

184

  cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);

294,646✔

186

  if (pInfo->pRow != NULL) {

294,646✔

187

    taosMemoryFree(pInfo->pRow);

212,043✔

188

    pInfo->pRow = NULL;

212,043✔

191

  if (pInfo->pStartCondInfo != NULL) {

294,646✔

192

    filterFreeInfo(pInfo->pStartCondInfo);

294,646✔

193

    pInfo->pStartCondInfo = NULL;

294,646✔

196

  if (pInfo->pEndCondInfo != NULL) {

294,646✔

197

    filterFreeInfo(pInfo->pEndCondInfo);

294,646✔

198

    pInfo->pEndCondInfo = NULL;

294,646✔

201

  cleanupBasicInfo(&pInfo->binfo);

294,646✔

202

  colDataDestroy(&pInfo->twAggSup.timeWindowData);

294,646✔

203

  pInfo->pOperator = NULL;

294,646✔

204

  cleanupAggSup(&pInfo->aggSup);

294,646✔

205

  cleanupExprSupp(&pInfo->scalarSup);

294,256✔

206

  taosMemoryFreeClear(param);

294,646✔

209

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

650,037✔

210

  int32_t                   code = TSDB_CODE_SUCCESS;

650,037✔

211

  int32_t                   lino = 0;

650,037✔

212

  SEventWindowOperatorInfo* pInfo = pOperator->info;

650,037✔

213

  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;

650,037✔

215

  SExprSupp* pSup = &pOperator->exprSupp;

650,037✔

216

  int32_t    order = pInfo->binfo.inputTsOrder;

650,037✔

218

  SSDataBlock* pRes = pInfo->binfo.pRes;

650,037✔

220

  blockDataCleanup(pRes);

650,037✔

222

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

650,037✔

223

  while (1) {

2,989,988✔

224

    SSDataBlock* pBlock = NULL;

3,640,025✔

225

    if (pInfo->pPreDataBlock == NULL) {

3,640,025✔

226

      pBlock = getNextBlockFromDownstream(pOperator, 0);

3,368,195✔

228

      pBlock = pInfo->pPreDataBlock;

271,830✔

229

      pInfo->pPreDataBlock = NULL;

271,830✔

232

    if (pBlock == NULL) {

3,640,025✔

233

      break;

309,120✔

236

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

3,330,905✔

237

    pRes->info.dataLoad = 1;

3,330,905✔

238

    code = setInputDataBlock(pSup, pBlock, order, pBlock->info.scanFlag, true);

3,330,905✔

239

    QUERY_CHECK_CODE(code, lino, _end);

3,330,905✔

241

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

3,330,905✔

242

    QUERY_CHECK_CODE(code, lino, _end);

3,330,905✔

245

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

3,330,905✔

246

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

55,699✔

247

                                   pInfo->scalarSup.numOfExprs, NULL, GET_STM_RTINFO(pOperator->pTaskInfo));

55,699✔

248

      QUERY_CHECK_CODE(code, lino, _end);

55,699✔

251

    code = eventWindowAggImpl(pOperator, pInfo, pBlock);

3,316,460✔

252

    QUERY_CHECK_CODE(code, lino, _end);

3,316,460✔

254

    code = doFilter(pRes, pSup->pFilterInfo, NULL, NULL);

3,303,344✔

255

    QUERY_CHECK_CODE(code, lino, _end);

3,303,344✔

257

    if (pRes->info.rows >= pOperator->resultInfo.threshold ||

3,303,344✔

258

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

3,234,470✔

259

      (*ppRes) = pRes;

313,356✔

260

      return code;

313,356✔

264

_end:

336,681✔

265

  if (code != TSDB_CODE_SUCCESS) {

336,681✔

266

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

27,561✔

267

    pTaskInfo->code = code;

27,561✔

268

    T_LONG_JMP(pTaskInfo->env, code);

27,561✔

270

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

309,120✔

271

  return code;

309,120✔

274

static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,

490,316,851✔

276

  if (*pResult == NULL) {

490,316,851✔

277

    SResultRow* p = taosMemoryCalloc(1, pAggSup->resultRowSize);

213,537✔

278

    if (!p) {

213,537✔

281

    pResultRowInfo->cur = (SResultRowPosition){.pageId = p->pageId, .offset = p->offset};

213,537✔

282

    *pResult = p;

213,537✔

285

  (*pResult)->win = *win;

490,316,851✔

287

  return setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);

490,316,461✔

290

static int32_t doEventWindowAggImpl(SEventWindowOperatorInfo* pInfo, SExprSupp* pSup, int32_t startIndex,

490,316,851✔

293

  int32_t code = TSDB_CODE_SUCCESS;

490,316,851✔

294

  SWindowRowsSup* pRowSup = &pInfo->winSup;

490,316,851✔

296

  int32_t numOfOutput = pSup->numOfExprs;

490,316,851✔

297

  int32_t numOfRows = endIndex - startIndex + 1;

490,316,851✔

299

  doKeepTuple(pRowSup, tsList[endIndex], endIndex, pBlock->info.id.groupId);

490,316,851✔

301

  code = setSingleOutputTupleBufv1(&pInfo->binfo.resultRowInfo, &pRowSup->win, &pInfo->pRow, pSup, &pInfo->aggSup);

490,316,461✔

302

  if (code != TSDB_CODE_SUCCESS) {  // null data, too many state code

490,316,851✔

307

  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, 0);

490,316,851✔

308

  pInfo->pRow->nOrigRows += numOfRows;

490,316,461✔

309

  code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startIndex, numOfRows,

490,316,851✔

310

                                         pBlock->info.rows, numOfOutput);

490,316,851✔

311

  return code;

490,316,851✔

314

int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {

3,316,460✔

315

  int32_t          code = TSDB_CODE_SUCCESS;

3,316,460✔

316

  int32_t          lino = 0;

3,316,460✔

317

  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;

3,316,460✔

318

  SExprSupp*       pSup = &pOperator->exprSupp;

3,316,460✔

319

  SSDataBlock*     pRes = pInfo->binfo.pRes;

3,316,460✔

320

  int64_t          gid = pBlock->info.id.groupId;

3,316,460✔

321

  SColumnInfoData *ps = NULL, *pe = NULL;

3,316,460✔

322

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

3,316,460✔

323

  QUERY_CHECK_NULL(pColInfoData, code, lino, _return, terrno);

3,316,460✔

324

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

3,316,460✔

325

  SWindowRowsSup*  pRowSup = &pInfo->winSup;

3,316,460✔

326

  int32_t          rowIndex = 0;

3,316,460✔

327

  STrueForInfo*    pTrueForInfo = getTrueForInfo(pOperator);

3,316,460✔

329

  pRowSup->numOfRows = 0;

3,316,460✔

330

  if (pInfo->groupId == 0) {

3,316,460✔

331

    pInfo->groupId = gid;

2,744,680✔

332

  } else if (pInfo->groupId != gid) {

571,780✔

334

    pInfo->inWindow = false;

290,590✔

335

    pInfo->groupId = gid;

290,590✔

336

    pInfo->winSup.lastTs = INT64_MIN;

290,590✔

337

    pInfo->pPreDataBlock = pBlock;

290,590✔

338

    goto _return;

290,590✔

340

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

3,025,870✔

342

  SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};

3,025,870✔

344

  code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &param1);

3,025,870✔

345

  QUERY_CHECK_CODE(code, lino, _return);

3,025,870✔

347

  int32_t status1 = 0;

3,025,870✔

348

  code = filterExecute(pInfo->pStartCondInfo, pBlock, &ps, NULL, param1.numOfCols, &status1);

3,025,870✔

349

  QUERY_CHECK_CODE(code, lino, _return);

3,025,870✔

351

  SFilterColumnParam param2 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};

3,025,870✔

352

  code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &param2);

3,025,870✔

353

  QUERY_CHECK_CODE(code, lino, _return);

3,025,870✔

355

  int32_t status2 = 0;

3,025,870✔

356

  code = filterExecute(pInfo->pEndCondInfo, pBlock, &pe, NULL, param2.numOfCols, &status2);

3,025,870✔

357

  QUERY_CHECK_CODE(code, lino, _return);

3,025,870✔

359

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

2,147,483,647✔

360

    if (pBlock->info.scanFlag != PRE_SCAN) {

2,147,483,647✔

361

      if (pInfo->winSup.lastTs == INT64_MIN) {

2,147,483,647✔

362

        pInfo->winSup.lastTs = tsList[i];

516,971✔

364

        if (tsList[i] == pInfo->winSup.lastTs) {

2,147,483,647✔

365

          qError("duplicate timestamp found in event window operator, groupId: %" PRId64 ", timestamp: %" PRId64,

13,116✔

367

          code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;

13,116✔

368

          QUERY_CHECK_CODE(code, lino, _return);

13,116✔

370

          pInfo->winSup.lastTs = tsList[i];

2,147,483,647✔

375

  int32_t startIndex = pInfo->inWindow ? 0 : -1;

3,012,364✔

376

  while (rowIndex < pBlock->info.rows) {

981,881,061✔

377

    if (pInfo->inWindow) {  // let's find the first end value

979,462,862✔

378

      for (rowIndex = startIndex; rowIndex < pBlock->info.rows; ++rowIndex) {

2,147,483,647✔

379

        if (((bool*)pe->pData)[rowIndex]) {

2,147,483,647✔

380

          break;

488,320,689✔

384

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

490,316,461✔

385

        code = doEventWindowAggImpl(pInfo, pSup, startIndex, rowIndex, pBlock, tsList, pTaskInfo);

488,320,689✔

386

        QUERY_CHECK_CODE(code, lino, _return);

488,321,079✔

387

        doUpdateNumOfRows(pSup->pCtx, pInfo->pRow, pSup->numOfExprs, pSup->rowEntryInfoOffset);

488,321,079✔

389

        if (!isTrueForSatisfied(pTrueForInfo, pRowSup->win.skey, pRowSup->win.ekey, pInfo->pRow->nOrigRows)) {

488,320,689✔

390

          qDebug("skip small window, groupId: %" PRId64 ", skey: %" PRId64 ", ekey: %" PRId64 ", nrows: %u",

133,937,640✔

394

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

354,383,049✔

395

            int32_t newSize = pRes->info.rows + pInfo->pRow->numOfRows;

675,360✔

396

            code = blockDataEnsureCapacity(pRes, newSize);

675,360✔

397

            QUERY_CHECK_CODE(code, lino, _return);

675,360✔

400

          code = copyResultrowToDataBlock(pSup->pExprInfo, pSup->numOfExprs, pInfo->pRow, pSup->pCtx, pRes,

354,383,049✔

401

                                          pSup->rowEntryInfoOffset, pTaskInfo);

354,383,049✔

402

          QUERY_CHECK_CODE(code, lino, _return);

354,383,439✔

404

          pRes->info.rows += pInfo->pRow->numOfRows;

354,383,439✔

406

        pInfo->pRow->numOfRows = 0;

488,320,689✔

407

        pInfo->pRow->nOrigRows = 0;

488,321,079✔

409

        pInfo->inWindow = false;

488,321,079✔

410

        rowIndex += 1;

488,321,079✔

412

        code = doEventWindowAggImpl(pInfo, pSup, startIndex, pBlock->info.rows - 1, pBlock, tsList, pTaskInfo);

1,995,772✔

413

        QUERY_CHECK_CODE(code, lino, _return);

1,995,772✔

416

      for (; rowIndex < pBlock->info.rows; ++rowIndex) {

1,314,934,050✔

417

        if (((bool*)ps->pData)[rowIndex]) {

1,314,339,885✔

418

          doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);

488,552,626✔

419

          pInfo->inWindow = true;

488,552,236✔

420

          startIndex = rowIndex;

488,552,236✔

421

          if (pInfo->pRow != NULL) {

488,552,236✔

422

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

488,339,089✔

424

          break;

488,551,846✔

428

      if (pInfo->inWindow) {

489,146,011✔

429

        continue;  // try to find the end position

488,551,846✔

431

        break;  // no valid start position, quit

594,555✔

436

_return:

3,316,460✔

438

  if (code != TSDB_CODE_SUCCESS) {

3,316,460✔

439

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

13,116✔

441

  colDataDestroy(ps);

3,316,460✔

442

  taosMemoryFree(ps);

3,316,460✔

443

  colDataDestroy(pe);

3,316,460✔

444

  taosMemoryFree(pe);

3,316,460✔

446

  return code;

3,316,460✔

taosdata / TDengine / #4979

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

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "executorInt.h"
17	#include "filter.h"
18	#include "function.h"
19	#include "functionMgt.h"
20	#include "operator.h"
21	#include "querytask.h"
22	#include "tcommon.h"
23	#include "tcompare.h"
24	#include "tdatablock.h"
25	#include "ttime.h"
26
27	static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** pRes);
28	static void destroyEWindowOperatorInfo(void* param);
29	static int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock);
30	void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo);
31
32	static int32_t resetEventWindowOperState(SOperatorInfo* pOper) {	1,494✔
33	SEventWindowOperatorInfo* pEvent = pOper->info;	1,494✔
34	SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;	1,494✔
35	SEventWinodwPhysiNode* pPhynode = (SEventWinodwPhysiNode*)pOper->pPhyNode;	1,494✔
36	pOper->status = OP_NOT_OPENED;	1,494✔
37
38	resetBasicOperatorState(&pEvent->binfo);	1,494✔
39	taosMemoryFreeClear(pEvent->pRow);	1,494✔
40
41	pEvent->groupId = 0;	1,494✔
42	pEvent->pPreDataBlock = NULL;	1,494✔
43	pEvent->inWindow = false;	1,494✔
44	pEvent->winSup.lastTs = INT64_MIN;	1,494✔
45
46	colDataDestroy(&pEvent->twAggSup.timeWindowData);	1,494✔
47	int32_t code = initExecTimeWindowInfo(&pEvent->twAggSup.timeWindowData, &pTaskInfo->window);	1,494✔
48	cleanupResultInfoInEventWindow(pOper, pEvent);	1,494✔
49
50	if (code == 0) {	1,494✔
51	code = resetAggSup(&pOper->exprSupp, &pEvent->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,	2,988✔
52	sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, pTaskInfo->streamInfo.pState,	1,494✔
53	&pTaskInfo->storageAPI.functionStore);
54	}
55	if (code == 0) {	1,494✔
56	code = resetExprSupp(&pEvent->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,	1,494✔
57	&pTaskInfo->storageAPI.functionStore);
58	}
59	return code;	1,494✔
60	}
61
62	int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,	294,646✔
63	SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
64	QRY_PARAM_CHECK(pOptrInfo);	294,646✔
65
66	int32_t code = TSDB_CODE_SUCCESS;	294,646✔
67	int32_t lino = 0;	294,646✔
68	SEventWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SEventWindowOperatorInfo));	294,646✔
69	SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));	294,256✔
70	if (pInfo == NULL \|\| pOperator == NULL) {	294,256✔
71	code = terrno;	×
72	goto _error;	×
73	}
74
75	pOperator->pPhyNode = physiNode;	294,256✔
76	pOperator->exprSupp.hasWindowOrGroup = true;	294,256✔
77	pOperator->exprSupp.hasWindow = true;	294,256✔
78
79	SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;	294,256✔
80
81	int32_t tsSlotId = ((SColumnNode*)pEventWindowNode->window.pTspk)->slotId;	294,256✔
82	code = filterInitFromNode((SNode*)pEventWindowNode->pStartCond, &pInfo->pStartCondInfo, 0,	294,646✔
83	pTaskInfo->pStreamRuntimeInfo);	294,646✔
84	QUERY_CHECK_CODE(code, lino, _error);	294,646✔
85
86	code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0,	294,646✔
87	pTaskInfo->pStreamRuntimeInfo);	294,646✔
88	QUERY_CHECK_CODE(code, lino, _error);	294,646✔
89
90	if (pEventWindowNode->window.pExprs != NULL) {	294,646✔
91	int32_t numOfScalarExpr = 0;	63,959✔
92	SExprInfo* pScalarExprInfo = NULL;	63,959✔
93
94	code = createExprInfo(pEventWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);	63,959✔
95	QUERY_CHECK_CODE(code, lino, _error);	63,959✔
96	code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);	63,959✔
97	QUERY_CHECK_CODE(code, lino, _error);	63,959✔
98	}
99
100	code = filterInitFromNode((SNode*)pEventWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,	294,646✔
101	pTaskInfo->pStreamRuntimeInfo);	294,646✔
102	QUERY_CHECK_CODE(code, lino, _error);	294,646✔
103
104	size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;	294,646✔
105
106	int32_t num = 0;	294,646✔
107	SExprInfo* pExprInfo = NULL;	294,646✔
108	code = createExprInfo(pEventWindowNode->window.pFuncs, NULL, &pExprInfo, &num);	294,646✔
109	QUERY_CHECK_CODE(code, lino, _error);	294,646✔
110
111	initResultSizeInfo(&pOperator->resultInfo, 4096);	294,646✔
112
113	code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,	589,292✔
114	pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);	294,646✔
115	QUERY_CHECK_CODE(code, lino, _error);	294,646✔
116
117	SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc);	294,646✔
118	QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);	294,646✔
119	initBasicInfo(&pInfo->binfo, pResBlock);	294,646✔
120
121	code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);	294,646✔
122	QUERY_CHECK_CODE(code, lino, _error);	294,646✔
123
124	initResultRowInfo(&pInfo->binfo.resultRowInfo);	294,646✔
125	pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;	294,646✔
126	pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;	294,646✔
127	pInfo->winSup.lastTs = INT64_MIN;	294,646✔
128
129	code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);	294,646✔
130	QUERY_CHECK_CODE(code, lino, _error);	294,646✔
131
132	pInfo->tsSlotId = tsSlotId;	294,646✔
133	pInfo->pPreDataBlock = NULL;	294,646✔
134	pInfo->pOperator = pOperator;	294,646✔
135	pInfo->trueForInfo.trueForType = pEventWindowNode->trueForType;	294,646✔
136	pInfo->trueForInfo.count = pEventWindowNode->trueForCount;	294,646✔
137	pInfo->trueForInfo.duration = pEventWindowNode->trueForDuration;	294,646✔
138
139	setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT, true, OP_NOT_OPENED, pInfo,	294,646✔
140	pTaskInfo);
141	pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, eventWindowAggregateNext, NULL, destroyEWindowOperatorInfo,	294,646✔
142	optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
143
144	setOperatorResetStateFn(pOperator, resetEventWindowOperState);	294,646✔
145	code = appendDownstream(pOperator, &downstream, 1);	294,646✔
146	if (code != TSDB_CODE_SUCCESS) {	294,646✔
147	goto _error;	×
148	}
149
150	*pOptrInfo = pOperator;	294,646✔
151	return TSDB_CODE_SUCCESS;	294,646✔
152
153	_error:	×
154	if (pInfo != NULL) {	×
155	destroyEWindowOperatorInfo(pInfo);	×
156	}
157
158	destroyOperatorAndDownstreams(pOperator, &downstream, 1);	×
159	pTaskInfo->code = code;	×
160	return code;	×
161	}
162
163	void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo) {	296,140✔
164	if (pInfo == NULL \|\| pInfo->pRow == NULL \|\| pOperator == NULL) {	296,140✔
165	return;	84,097✔
166	}
167	SExprSupp* pSup = &pOperator->exprSupp;	212,043✔
168	for (int32_t j = 0; j < pSup->numOfExprs; ++j) {	212,043✔
169	pSup->pCtx[j].resultInfo = getResultEntryInfo(pInfo->pRow, j, pSup->rowEntryInfoOffset);	×
170	if (pSup->pCtx[j].fpSet.cleanup) {	×
171	pSup->pCtx[j].fpSet.cleanup(&pSup->pCtx[j]);	×
172	}
173	}
174	}
175
176	void destroyEWindowOperatorInfo(void* param) {	294,646✔
177	SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;	294,646✔
178	if (pInfo == NULL) {	294,646✔
179	return;	×
180	}
181
182	// First cleanup function contexts that may reference result buffers/state.
183	// This must happen before freeing any buffers that those cleanups might touch.
184	cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);	294,646✔
185
186	if (pInfo->pRow != NULL) {	294,646✔
187	taosMemoryFree(pInfo->pRow);	212,043✔
188	pInfo->pRow = NULL;	212,043✔
189	}
190
191	if (pInfo->pStartCondInfo != NULL) {	294,646✔
192	filterFreeInfo(pInfo->pStartCondInfo);	294,646✔
193	pInfo->pStartCondInfo = NULL;	294,646✔
194	}
195
196	if (pInfo->pEndCondInfo != NULL) {	294,646✔
197	filterFreeInfo(pInfo->pEndCondInfo);	294,646✔
198	pInfo->pEndCondInfo = NULL;	294,646✔
199	}
200
201	cleanupBasicInfo(&pInfo->binfo);	294,646✔
202	colDataDestroy(&pInfo->twAggSup.timeWindowData);	294,646✔
203	pInfo->pOperator = NULL;	294,646✔
204	cleanupAggSup(&pInfo->aggSup);	294,646✔
205	cleanupExprSupp(&pInfo->scalarSup);	294,256✔
206	taosMemoryFreeClear(param);	294,646✔
207	}
208
209	static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {	650,037✔
210	int32_t code = TSDB_CODE_SUCCESS;	650,037✔
211	int32_t lino = 0;	650,037✔
212	SEventWindowOperatorInfo* pInfo = pOperator->info;	650,037✔
213	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	650,037✔
214
215	SExprSupp* pSup = &pOperator->exprSupp;	650,037✔
216	int32_t order = pInfo->binfo.inputTsOrder;	650,037✔
217
218	SSDataBlock* pRes = pInfo->binfo.pRes;	650,037✔
219
220	blockDataCleanup(pRes);	650,037✔
221
222	SOperatorInfo* downstream = pOperator->pDownstream[0];	650,037✔
223	while (1) {	2,989,988✔
224	SSDataBlock* pBlock = NULL;	3,640,025✔
225	if (pInfo->pPreDataBlock == NULL) {	3,640,025✔
226	pBlock = getNextBlockFromDownstream(pOperator, 0);	3,368,195✔
227	} else {
228	pBlock = pInfo->pPreDataBlock;	271,830✔
229	pInfo->pPreDataBlock = NULL;	271,830✔
230	}
231
232	if (pBlock == NULL) {	3,640,025✔
233	break;	309,120✔
234	}
235
236	pRes->info.scanFlag = pBlock->info.scanFlag;	3,330,905✔
237	pRes->info.dataLoad = 1;	3,330,905✔
238	code = setInputDataBlock(pSup, pBlock, order, pBlock->info.scanFlag, true);	3,330,905✔
239	QUERY_CHECK_CODE(code, lino, _end);	3,330,905✔
240
241	code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);	3,330,905✔
242	QUERY_CHECK_CODE(code, lino, _end);	3,330,905✔
243
244	// there is an scalar expression that needs to be calculated right before apply the group aggregation.
245	if (pInfo->scalarSup.pExprInfo != NULL) {	3,330,905✔
246	code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,	55,699✔
247	pInfo->scalarSup.numOfExprs, NULL, GET_STM_RTINFO(pOperator->pTaskInfo));	55,699✔
248	QUERY_CHECK_CODE(code, lino, _end);	55,699✔
249	}
250
251	code = eventWindowAggImpl(pOperator, pInfo, pBlock);	3,316,460✔
252	QUERY_CHECK_CODE(code, lino, _end);	3,316,460✔
253
254	code = doFilter(pRes, pSup->pFilterInfo, NULL, NULL);	3,303,344✔
255	QUERY_CHECK_CODE(code, lino, _end);	3,303,344✔
256
257	if (pRes->info.rows >= pOperator->resultInfo.threshold \|\|	3,303,344✔
258	(pRes->info.id.groupId != pInfo->groupId && pRes->info.rows > 0)) {	3,234,470✔
259	(*ppRes) = pRes;	313,356✔
260	return code;	313,356✔
261	}
262	}
263
264	_end:	336,681✔
265	if (code != TSDB_CODE_SUCCESS) {	336,681✔
266	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	27,561✔
267	pTaskInfo->code = code;	27,561✔
268	T_LONG_JMP(pTaskInfo->env, code);	27,561✔
269	}
270	(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;	309,120✔
271	return code;	309,120✔
272	}
273
274	static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,	490,316,851✔
275	SExprSupp* pExprSup, SAggSupporter* pAggSup) {
276	if (*pResult == NULL) {	490,316,851✔
277	SResultRow* p = taosMemoryCalloc(1, pAggSup->resultRowSize);	213,537✔
278	if (!p) {	213,537✔
279	return terrno;	×
280	}
281	pResultRowInfo->cur = (SResultRowPosition){.pageId = p->pageId, .offset = p->offset};	213,537✔
282	*pResult = p;	213,537✔
283	}
284
285	(pResult)->win = win;	490,316,851✔
286
287	return setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	490,316,461✔
288	}
289
290	static int32_t doEventWindowAggImpl(SEventWindowOperatorInfo* pInfo, SExprSupp* pSup, int32_t startIndex,	490,316,851✔
291	int32_t endIndex, const SSDataBlock* pBlock, int64_t* tsList,
292	SExecTaskInfo* pTaskInfo) {
293	int32_t code = TSDB_CODE_SUCCESS;	490,316,851✔
294	SWindowRowsSup* pRowSup = &pInfo->winSup;	490,316,851✔
295
296	int32_t numOfOutput = pSup->numOfExprs;	490,316,851✔
297	int32_t numOfRows = endIndex - startIndex + 1;	490,316,851✔
298
299	doKeepTuple(pRowSup, tsList[endIndex], endIndex, pBlock->info.id.groupId);	490,316,851✔
300
301	code = setSingleOutputTupleBufv1(&pInfo->binfo.resultRowInfo, &pRowSup->win, &pInfo->pRow, pSup, &pInfo->aggSup);	490,316,461✔
302	if (code != TSDB_CODE_SUCCESS) { // null data, too many state code	490,316,851✔
303	qError("failed to set single output tuple buffer, code:%d", code);	×
304	return code;	×
305	}
306
307	updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, 0);	490,316,851✔
308	pInfo->pRow->nOrigRows += numOfRows;	490,316,461✔
309	code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startIndex, numOfRows,	490,316,851✔
310	pBlock->info.rows, numOfOutput);	490,316,851✔
311	return code;	490,316,851✔
312	}
313
314	int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {	3,316,460✔
315	int32_t code = TSDB_CODE_SUCCESS;	3,316,460✔
316	int32_t lino = 0;	3,316,460✔
317	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	3,316,460✔
318	SExprSupp* pSup = &pOperator->exprSupp;	3,316,460✔
319	SSDataBlock* pRes = pInfo->binfo.pRes;	3,316,460✔
320	int64_t gid = pBlock->info.id.groupId;	3,316,460✔
321	SColumnInfoData ps = NULL, pe = NULL;	3,316,460✔
322	SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);	3,316,460✔
323	QUERY_CHECK_NULL(pColInfoData, code, lino, _return, terrno);	3,316,460✔
324	TSKEY* tsList = (TSKEY*)pColInfoData->pData;	3,316,460✔
325	SWindowRowsSup* pRowSup = &pInfo->winSup;	3,316,460✔
326	int32_t rowIndex = 0;	3,316,460✔
327	STrueForInfo* pTrueForInfo = getTrueForInfo(pOperator);	3,316,460✔
328
329	pRowSup->numOfRows = 0;	3,316,460✔
330	if (pInfo->groupId == 0) {	3,316,460✔
331	pInfo->groupId = gid;	2,744,680✔
332	} else if (pInfo->groupId != gid) {	571,780✔
333	// this is a new group, reset the info
334	pInfo->inWindow = false;	290,590✔
335	pInfo->groupId = gid;	290,590✔
336	pInfo->winSup.lastTs = INT64_MIN;	290,590✔
337	pInfo->pPreDataBlock = pBlock;	290,590✔
338	goto _return;	290,590✔
339	}
340	pRes->info.id.groupId = pInfo->groupId;	3,025,870✔
341
342	SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	3,025,870✔
343
344	code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &param1);	3,025,870✔
345	QUERY_CHECK_CODE(code, lino, _return);	3,025,870✔
346
347	int32_t status1 = 0;	3,025,870✔
348	code = filterExecute(pInfo->pStartCondInfo, pBlock, &ps, NULL, param1.numOfCols, &status1);	3,025,870✔
349	QUERY_CHECK_CODE(code, lino, _return);	3,025,870✔
350
351	SFilterColumnParam param2 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	3,025,870✔
352	code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &param2);	3,025,870✔
353	QUERY_CHECK_CODE(code, lino, _return);	3,025,870✔
354
355	int32_t status2 = 0;	3,025,870✔
356	code = filterExecute(pInfo->pEndCondInfo, pBlock, &pe, NULL, param2.numOfCols, &status2);	3,025,870✔
357	QUERY_CHECK_CODE(code, lino, _return);	3,025,870✔
358
359	for (int32_t i = 0; i < pBlock->info.rows; ++i) {	2,147,483,647✔
360	if (pBlock->info.scanFlag != PRE_SCAN) {	2,147,483,647✔
361	if (pInfo->winSup.lastTs == INT64_MIN) {	2,147,483,647✔
362	pInfo->winSup.lastTs = tsList[i];	516,971✔
363	} else {
364	if (tsList[i] == pInfo->winSup.lastTs) {	2,147,483,647✔
365	qError("duplicate timestamp found in event window operator, groupId: %" PRId64 ", timestamp: %" PRId64,	13,116✔
366	gid, tsList[i]);
367	code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;	13,116✔
368	QUERY_CHECK_CODE(code, lino, _return);	13,116✔
369	} else {
370	pInfo->winSup.lastTs = tsList[i];	2,147,483,647✔
371	}
372	}
373	}
374	}
375	int32_t startIndex = pInfo->inWindow ? 0 : -1;	3,012,364✔
376	while (rowIndex < pBlock->info.rows) {	981,881,061✔
377	if (pInfo->inWindow) { // let's find the first end value	979,462,862✔
378	for (rowIndex = startIndex; rowIndex < pBlock->info.rows; ++rowIndex) {	2,147,483,647✔
379	if (((bool*)pe->pData)[rowIndex]) {	2,147,483,647✔
380	break;	488,320,689✔
381	}
382	}
383
384	if (rowIndex < pBlock->info.rows) {	490,316,461✔
385	code = doEventWindowAggImpl(pInfo, pSup, startIndex, rowIndex, pBlock, tsList, pTaskInfo);	488,320,689✔
386	QUERY_CHECK_CODE(code, lino, _return);	488,321,079✔
387	doUpdateNumOfRows(pSup->pCtx, pInfo->pRow, pSup->numOfExprs, pSup->rowEntryInfoOffset);	488,321,079✔
388
389	if (!isTrueForSatisfied(pTrueForInfo, pRowSup->win.skey, pRowSup->win.ekey, pInfo->pRow->nOrigRows)) {	488,320,689✔
390	qDebug("skip small window, groupId: %" PRId64 ", skey: %" PRId64 ", ekey: %" PRId64 ", nrows: %u",	133,937,640✔
391	pInfo->groupId, pRowSup->win.skey, pRowSup->win.ekey, pInfo->pRow->nOrigRows);
392	} else {
393	// check buffer size
394	if (pRes->info.rows + pInfo->pRow->numOfRows >= pRes->info.capacity) {	354,383,049✔
395	int32_t newSize = pRes->info.rows + pInfo->pRow->numOfRows;	675,360✔
396	code = blockDataEnsureCapacity(pRes, newSize);	675,360✔
397	QUERY_CHECK_CODE(code, lino, _return);	675,360✔
398	}
399
400	code = copyResultrowToDataBlock(pSup->pExprInfo, pSup->numOfExprs, pInfo->pRow, pSup->pCtx, pRes,	354,383,049✔
401	pSup->rowEntryInfoOffset, pTaskInfo);	354,383,049✔
402	QUERY_CHECK_CODE(code, lino, _return);	354,383,439✔
403
404	pRes->info.rows += pInfo->pRow->numOfRows;	354,383,439✔
405	}
406	pInfo->pRow->numOfRows = 0;	488,320,689✔
407	pInfo->pRow->nOrigRows = 0;	488,321,079✔
408
409	pInfo->inWindow = false;	488,321,079✔
410	rowIndex += 1;	488,321,079✔
411	} else {
412	code = doEventWindowAggImpl(pInfo, pSup, startIndex, pBlock->info.rows - 1, pBlock, tsList, pTaskInfo);	1,995,772✔
413	QUERY_CHECK_CODE(code, lino, _return);	1,995,772✔
414	}
415	} else { // find the first start value that is fulfill for the start condition
416	for (; rowIndex < pBlock->info.rows; ++rowIndex) {	1,314,934,050✔
417	if (((bool*)ps->pData)[rowIndex]) {	1,314,339,885✔
418	doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);	488,552,626✔
419	pInfo->inWindow = true;	488,552,236✔
420	startIndex = rowIndex;	488,552,236✔
421	if (pInfo->pRow != NULL) {	488,552,236✔
422	clearResultRowInitFlag(pSup->pCtx, pSup->numOfExprs);	488,339,089✔
423	}
424	break;	488,551,846✔
425	}
426	}
427
428	if (pInfo->inWindow) {	489,146,011✔
429	continue; // try to find the end position	488,551,846✔
430	} else {
431	break; // no valid start position, quit	594,555✔
432	}
433	}
434	}
435
436	_return:	3,316,460✔
437
438	if (code != TSDB_CODE_SUCCESS) {	3,316,460✔
439	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	13,116✔
440	}
441	colDataDestroy(ps);	3,316,460✔
442	taosMemoryFree(ps);	3,316,460✔
443	colDataDestroy(pe);	3,316,460✔
444	taosMemoryFree(pe);	3,316,460✔
445
446	return code;	3,316,460✔
447	}

taosdata / TDengine / #4979

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