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

32

static int32_t resetEventWindowOperState(SOperatorInfo* pOper) {

1,464✔

33

  SEventWindowOperatorInfo* pEvent = pOper->info;

1,464✔

34

  SExecTaskInfo*           pTaskInfo = pOper->pTaskInfo;

1,464✔

35

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

1,464✔

36

  pOper->status = OP_NOT_OPENED;

1,464✔

38

  resetBasicOperatorState(&pEvent->binfo);

1,464✔

39

  taosMemoryFreeClear(pEvent->pRow);

1,464✔

41

  pEvent->groupId = 0;

1,464✔

42

  pEvent->pPreDataBlock = NULL;

1,464✔

43

  pEvent->inWindow = false;

1,464✔

44

  pEvent->winSup.lastTs = INT64_MIN;

1,464✔

46

  colDataDestroy(&pEvent->twAggSup.timeWindowData);

1,464✔

47

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

1,464✔

48

  cleanupResultInfoInEventWindow(pOper, pEvent);

1,464✔

50

  if (code == 0) {

1,464✔

51

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

2,928✔

52

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

1,464✔

55

  if (code == 0) {

1,464✔

56

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

1,464✔

59

  return code;

1,464✔

62

int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,

2,001,391✔

64

  QRY_PARAM_CHECK(pOptrInfo);

2,001,391✔

66

  int32_t                   code = TSDB_CODE_SUCCESS;

2,001,391✔

67

  int32_t                   lino = 0;

2,001,391✔

68

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

2,001,391✔

69

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

2,001,391✔

70

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

2,001,391✔

75

  pOperator->pPhyNode = physiNode;

2,001,391✔

76

  pOperator->exprSupp.hasWindowOrGroup = true;

2,001,391✔

77

  pOperator->exprSupp.hasWindow = true;

2,001,391✔

79

  SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;

2,001,391✔

81

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

2,001,391✔

82

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

2,001,391✔

83

                            pTaskInfo->pStreamRuntimeInfo);

2,001,391✔

84

  QUERY_CHECK_CODE(code, lino, _error);

2,001,391✔

86

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

2,001,391✔

87

                            pTaskInfo->pStreamRuntimeInfo);

2,001,391✔

88

  QUERY_CHECK_CODE(code, lino, _error);

2,001,391✔

90

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

2,001,391✔

91

    int32_t    numOfScalarExpr = 0;

1,799,344✔

92

    SExprInfo* pScalarExprInfo = NULL;

1,799,344✔

94

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

1,799,344✔

95

    QUERY_CHECK_CODE(code, lino, _error);

1,799,344✔

96

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

1,799,344✔

97

    QUERY_CHECK_CODE(code, lino, _error);

1,799,344✔

100

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

2,001,391✔

101

                            pTaskInfo->pStreamRuntimeInfo);

2,001,391✔

102

  QUERY_CHECK_CODE(code, lino, _error);

2,001,391✔

104

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

2,001,391✔

106

  int32_t    num = 0;

2,001,391✔

107

  SExprInfo* pExprInfo = NULL;

2,001,391✔

108

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

2,001,391✔

109

  QUERY_CHECK_CODE(code, lino, _error);

2,001,391✔

111

  initResultSizeInfo(&pOperator->resultInfo, 4096);

2,001,391✔

113

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

4,002,782✔

114

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

2,001,391✔

115

  QUERY_CHECK_CODE(code, lino, _error);

2,001,391✔

117

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

2,001,391✔

118

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

2,001,391✔

119

  initBasicInfo(&pInfo->binfo, pResBlock);

2,001,391✔

121

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

2,001,391✔

122

  QUERY_CHECK_CODE(code, lino, _error);

2,001,391✔

124

  initResultRowInfo(&pInfo->binfo.resultRowInfo);

2,001,391✔

125

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

2,001,391✔

126

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

2,001,391✔

127

  pInfo->winSup.lastTs = INT64_MIN;

2,001,391✔

129

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

2,001,391✔

130

  QUERY_CHECK_CODE(code, lino, _error);

2,001,391✔

132

  pInfo->tsSlotId = tsSlotId;

2,001,391✔

133

  pInfo->pPreDataBlock = NULL;

2,001,391✔

134

  pInfo->pOperator = pOperator;

2,001,391✔

135

  pInfo->trueForLimit = pEventWindowNode->trueForLimit;

2,001,391✔

137

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

2,001,391✔

139

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

2,001,391✔

142

  setOperatorResetStateFn(pOperator, resetEventWindowOperState);

2,001,391✔

143

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

2,001,391✔

144

  if (code != TSDB_CODE_SUCCESS) {

2,001,391✔

148

  *pOptrInfo = pOperator;

2,001,391✔

149

  return TSDB_CODE_SUCCESS;

2,001,391✔

161

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

2,002,855✔

162

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

2,002,855✔

163

    return;

1,168,342✔

165

  SExprSupp*       pSup = &pOperator->exprSupp;

834,513✔

166

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

3,224,061✔

167

    pSup->pCtx[j].resultInfo = getResultEntryInfo(pInfo->pRow, j, pSup->rowEntryInfoOffset);

2,389,548✔

168

    if (pSup->pCtx[j].fpSet.cleanup) {

2,389,548✔

174

void destroyEWindowOperatorInfo(void* param) {

2,001,391✔

175

  SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;

2,001,391✔

176

  if (pInfo == NULL) {

2,001,391✔

180

  if (pInfo->pRow != NULL) {

2,001,391✔

181

    taosMemoryFree(pInfo->pRow);

834,513✔

184

  if (pInfo->pStartCondInfo != NULL) {

2,001,391✔

185

    filterFreeInfo(pInfo->pStartCondInfo);

2,001,391✔

186

    pInfo->pStartCondInfo = NULL;

2,001,391✔

189

  if (pInfo->pEndCondInfo != NULL) {

2,001,391✔

190

    filterFreeInfo(pInfo->pEndCondInfo);

2,001,391✔

191

    pInfo->pEndCondInfo = NULL;

2,001,391✔

194

  cleanupBasicInfo(&pInfo->binfo);

2,001,391✔

195

  colDataDestroy(&pInfo->twAggSup.timeWindowData);

2,001,391✔

197

  cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);

2,001,391✔

198

  pInfo->pOperator = NULL;

2,001,391✔

199

  cleanupAggSup(&pInfo->aggSup);

2,001,391✔

200

  cleanupExprSupp(&pInfo->scalarSup);

2,001,391✔

201

  taosMemoryFreeClear(param);

2,001,391✔

204

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

2,734,026✔

205

  int32_t                   code = TSDB_CODE_SUCCESS;

2,734,026✔

206

  int32_t                   lino = 0;

2,734,026✔

207

  SEventWindowOperatorInfo* pInfo = pOperator->info;

2,734,026✔

208

  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;

2,734,026✔

210

  SExprSupp* pSup = &pOperator->exprSupp;

2,734,026✔

211

  int32_t    order = pInfo->binfo.inputTsOrder;

2,734,026✔

213

  SSDataBlock* pRes = pInfo->binfo.pRes;

2,734,026✔

215

  blockDataCleanup(pRes);

2,734,026✔

217

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

2,734,026✔

218

  while (1) {

2,433,672✔

219

    SSDataBlock* pBlock = NULL;

5,167,698✔

220

    if (pInfo->pPreDataBlock == NULL) {

5,167,698✔

221

      pBlock = getNextBlockFromDownstream(pOperator, 0);

4,930,203✔

223

      pBlock = pInfo->pPreDataBlock;

237,495✔

224

      pInfo->pPreDataBlock = NULL;

237,495✔

227

    if (pBlock == NULL) {

5,167,698✔

228

      break;

1,930,572✔

231

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

3,237,126✔

232

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

3,237,126✔

233

    QUERY_CHECK_CODE(code, lino, _end);

3,237,126✔

235

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

3,237,126✔

236

    QUERY_CHECK_CODE(code, lino, _end);

3,237,126✔

239

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

3,237,126✔

240

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

1,235,279✔

241

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

1,235,279✔

242

      QUERY_CHECK_CODE(code, lino, _end);

1,235,279✔

245

    code = eventWindowAggImpl(pOperator, pInfo, pBlock);

2,783,685✔

246

    QUERY_CHECK_CODE(code, lino, _end);

2,783,685✔

248

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

2,680,452✔

249

    QUERY_CHECK_CODE(code, lino, _end);

2,680,452✔

251

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

2,680,452✔

252

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

2,644,724✔

253

      (*ppRes) = pRes;

246,780✔

254

      return code;

246,780✔

258

_end:

2,487,246✔

259

  if (code != TSDB_CODE_SUCCESS) {

2,487,246✔

260

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

556,674✔

261

    pTaskInfo->code = code;

556,674✔

262

    T_LONG_JMP(pTaskInfo->env, code);

556,674✔

264

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

1,930,572✔

265

  return code;

1,930,572✔

268

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

236,858,148✔

270

  if (*pResult == NULL) {

236,858,148✔

271

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

835,977✔

272

    if (!p) {

835,977✔

275

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

835,977✔

276

    *pResult = p;

835,977✔

279

  (*pResult)->win = *win;

236,858,148✔

281

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

236,858,148✔

284

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

236,858,148✔

287

  int32_t code = TSDB_CODE_SUCCESS;

236,858,148✔

288

  SWindowRowsSup* pRowSup = &pInfo->winSup;

236,858,148✔

290

  int32_t numOfOutput = pSup->numOfExprs;

236,858,148✔

291

  int32_t numOfRows = endIndex - startIndex + 1;

236,858,148✔

293

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

236,858,148✔

295

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

236,858,148✔

296

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

236,858,148✔

301

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

236,858,148✔

302

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

236,858,148✔

303

                                         pBlock->info.rows, numOfOutput);

236,858,148✔

304

  return code;

236,858,148✔

307

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

2,783,685✔

308

  int32_t          code = TSDB_CODE_SUCCESS;

2,783,685✔

309

  int32_t          lino = 0;

2,783,685✔

310

  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;

2,783,685✔

311

  SExprSupp*       pSup = &pOperator->exprSupp;

2,783,685✔

312

  SSDataBlock*     pRes = pInfo->binfo.pRes;

2,783,685✔

313

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

2,783,685✔

314

  SColumnInfoData *ps = NULL, *pe = NULL;

2,783,685✔

315

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

2,783,685✔

316

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

2,783,685✔

317

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

2,783,685✔

318

  SWindowRowsSup*  pRowSup = &pInfo->winSup;

2,783,685✔

319

  int32_t          rowIndex = 0;

2,783,685✔

320

  int64_t          minWindowSize = getMinWindowSize(pOperator);

2,783,685✔

322

  pRowSup->numOfRows = 0;

2,783,685✔

323

  if (pInfo->groupId == 0) {

2,783,685✔

324

    pInfo->groupId = gid;

2,280,197✔

325

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

503,488✔

327

    pInfo->inWindow = false;

256,759✔

328

    pInfo->groupId = gid;

256,759✔

329

    pInfo->winSup.lastTs = INT64_MIN;

256,759✔

330

    pInfo->pPreDataBlock = pBlock;

256,759✔

331

    goto _return;

256,759✔

333

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

2,526,926✔

335

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

2,526,926✔

337

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

2,526,926✔

338

  QUERY_CHECK_CODE(code, lino, _return);

2,526,926✔

340

  int32_t status1 = 0;

2,526,926✔

341

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

2,526,926✔

342

  QUERY_CHECK_CODE(code, lino, _return);

2,526,926✔

344

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

2,526,926✔

345

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

2,526,926✔

346

  QUERY_CHECK_CODE(code, lino, _return);

2,526,926✔

348

  int32_t status2 = 0;

2,526,926✔

349

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

2,526,926✔

350

  QUERY_CHECK_CODE(code, lino, _return);

2,526,926✔

352

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

1,837,195,150✔

353

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

1,834,771,457✔

354

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

1,834,771,457✔

355

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

1,182,938✔

357

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

1,833,588,519✔

358

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

103,233✔

360

          code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;

103,233✔

361

          QUERY_CHECK_CODE(code, lino, _return);

103,233✔

363

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

1,833,485,286✔

368

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

2,423,693✔

369

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

475,283,551✔

370

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

473,308,337✔

371

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

1,352,217,881✔

372

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

1,350,970,687✔

373

          break;

235,610,954✔

377

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

236,858,148✔

378

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

235,610,954✔

379

        QUERY_CHECK_CODE(code, lino, _return);

235,610,954✔

380

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

235,610,954✔

382

        int64_t delta = pRowSup->win.ekey - pRowSup->win.skey;

235,610,954✔

383

        if(pInfo->binfo.inputTsOrder == ORDER_DESC) {

235,610,954✔

384

          delta = -delta;

1,512✔

386

        if (delta < minWindowSize) {

235,610,954✔

387

          qDebug("skip small window, groupId: %" PRId64 ", windowSize: %" PRId64 ", minWindowSize: %" PRId64,

1,860✔

391

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

235,609,094✔

392

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

666,432✔

393

            code = blockDataEnsureCapacity(pRes, newSize);

666,432✔

394

            QUERY_CHECK_CODE(code, lino, _return);

666,432✔

397

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

235,609,094✔

398

                                          pSup->rowEntryInfoOffset, pTaskInfo);

235,609,094✔

399

          QUERY_CHECK_CODE(code, lino, _return);

235,609,094✔

401

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

235,609,094✔

403

        pInfo->pRow->numOfRows = 0;

235,610,954✔

405

        pInfo->inWindow = false;

235,610,954✔

406

        rowIndex += 1;

235,610,954✔

408

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

1,247,194✔

409

        QUERY_CHECK_CODE(code, lino, _return);

1,247,194✔

412

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

720,044,493✔

413

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

719,596,014✔

414

          doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);

236,001,710✔

415

          pInfo->inWindow = true;

236,001,710✔

416

          startIndex = rowIndex;

236,001,710✔

417

          if (pInfo->pRow != NULL) {

236,001,710✔

418

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

235,165,733✔

420

          break;

236,001,710✔

424

      if (pInfo->inWindow) {

236,450,565✔

425

        continue;  // try to find the end position

236,001,710✔

427

        break;  // no valid start position, quit

448,855✔

432

_return:

2,783,685✔

434

  if (code != TSDB_CODE_SUCCESS) {

2,783,685✔

435

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

103,233✔

437

  colDataDestroy(ps);

2,783,685✔

438

  taosMemoryFree(ps);

2,783,685✔

439

  colDataDestroy(pe);

2,783,685✔

440

  taosMemoryFree(pe);

2,783,685✔

442

  return code;

2,783,685✔

taosdata / TDengine / #4932

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,464✔
33	SEventWindowOperatorInfo* pEvent = pOper->info;	1,464✔
34	SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;	1,464✔
35	SEventWinodwPhysiNode* pPhynode = (SEventWinodwPhysiNode*)pOper->pPhyNode;	1,464✔
36	pOper->status = OP_NOT_OPENED;	1,464✔
37
38	resetBasicOperatorState(&pEvent->binfo);	1,464✔
39	taosMemoryFreeClear(pEvent->pRow);	1,464✔
40
41	pEvent->groupId = 0;	1,464✔
42	pEvent->pPreDataBlock = NULL;	1,464✔
43	pEvent->inWindow = false;	1,464✔
44	pEvent->winSup.lastTs = INT64_MIN;	1,464✔
45
46	colDataDestroy(&pEvent->twAggSup.timeWindowData);	1,464✔
47	int32_t code = initExecTimeWindowInfo(&pEvent->twAggSup.timeWindowData, &pTaskInfo->window);	1,464✔
48	cleanupResultInfoInEventWindow(pOper, pEvent);	1,464✔
49
50	if (code == 0) {	1,464✔
51	code = resetAggSup(&pOper->exprSupp, &pEvent->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,	2,928✔
52	sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, pTaskInfo->streamInfo.pState,	1,464✔
53	&pTaskInfo->storageAPI.functionStore);
54	}
55	if (code == 0) {	1,464✔
56	code = resetExprSupp(&pEvent->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,	1,464✔
57	&pTaskInfo->storageAPI.functionStore);
58	}
59	return code;	1,464✔
60	}
61
62	int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,	2,001,391✔
63	SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
64	QRY_PARAM_CHECK(pOptrInfo);	2,001,391✔
65
66	int32_t code = TSDB_CODE_SUCCESS;	2,001,391✔
67	int32_t lino = 0;	2,001,391✔
68	SEventWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SEventWindowOperatorInfo));	2,001,391✔
69	SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));	2,001,391✔
70	if (pInfo == NULL \|\| pOperator == NULL) {	2,001,391✔
71	code = terrno;	×
72	goto _error;	×
73	}
74
75	pOperator->pPhyNode = physiNode;	2,001,391✔
76	pOperator->exprSupp.hasWindowOrGroup = true;	2,001,391✔
77	pOperator->exprSupp.hasWindow = true;	2,001,391✔
78
79	SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;	2,001,391✔
80
81	int32_t tsSlotId = ((SColumnNode*)pEventWindowNode->window.pTspk)->slotId;	2,001,391✔
82	code = filterInitFromNode((SNode*)pEventWindowNode->pStartCond, &pInfo->pStartCondInfo, 0,	2,001,391✔
83	pTaskInfo->pStreamRuntimeInfo);	2,001,391✔
84	QUERY_CHECK_CODE(code, lino, _error);	2,001,391✔
85
86	code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0,	2,001,391✔
87	pTaskInfo->pStreamRuntimeInfo);	2,001,391✔
88	QUERY_CHECK_CODE(code, lino, _error);	2,001,391✔
89
90	if (pEventWindowNode->window.pExprs != NULL) {	2,001,391✔
91	int32_t numOfScalarExpr = 0;	1,799,344✔
92	SExprInfo* pScalarExprInfo = NULL;	1,799,344✔
93
94	code = createExprInfo(pEventWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);	1,799,344✔
95	QUERY_CHECK_CODE(code, lino, _error);	1,799,344✔
96	code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);	1,799,344✔
97	QUERY_CHECK_CODE(code, lino, _error);	1,799,344✔
98	}
99
100	code = filterInitFromNode((SNode*)pEventWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,	2,001,391✔
101	pTaskInfo->pStreamRuntimeInfo);	2,001,391✔
102	QUERY_CHECK_CODE(code, lino, _error);	2,001,391✔
103
104	size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;	2,001,391✔
105
106	int32_t num = 0;	2,001,391✔
107	SExprInfo* pExprInfo = NULL;	2,001,391✔
108	code = createExprInfo(pEventWindowNode->window.pFuncs, NULL, &pExprInfo, &num);	2,001,391✔
109	QUERY_CHECK_CODE(code, lino, _error);	2,001,391✔
110
111	initResultSizeInfo(&pOperator->resultInfo, 4096);	2,001,391✔
112
113	code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,	4,002,782✔
114	pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);	2,001,391✔
115	QUERY_CHECK_CODE(code, lino, _error);	2,001,391✔
116
117	SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc);	2,001,391✔
118	QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);	2,001,391✔
119	initBasicInfo(&pInfo->binfo, pResBlock);	2,001,391✔
120
121	code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);	2,001,391✔
122	QUERY_CHECK_CODE(code, lino, _error);	2,001,391✔
123
124	initResultRowInfo(&pInfo->binfo.resultRowInfo);	2,001,391✔
125	pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;	2,001,391✔
126	pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;	2,001,391✔
127	pInfo->winSup.lastTs = INT64_MIN;	2,001,391✔
128
129	code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);	2,001,391✔
130	QUERY_CHECK_CODE(code, lino, _error);	2,001,391✔
131
132	pInfo->tsSlotId = tsSlotId;	2,001,391✔
133	pInfo->pPreDataBlock = NULL;	2,001,391✔
134	pInfo->pOperator = pOperator;	2,001,391✔
135	pInfo->trueForLimit = pEventWindowNode->trueForLimit;	2,001,391✔
136
137	setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT, true, OP_NOT_OPENED, pInfo,	2,001,391✔
138	pTaskInfo);
139	pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, eventWindowAggregateNext, NULL, destroyEWindowOperatorInfo,	2,001,391✔
140	optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
141
142	setOperatorResetStateFn(pOperator, resetEventWindowOperState);	2,001,391✔
143	code = appendDownstream(pOperator, &downstream, 1);	2,001,391✔
144	if (code != TSDB_CODE_SUCCESS) {	2,001,391✔
145	goto _error;	×
146	}
147
148	*pOptrInfo = pOperator;	2,001,391✔
149	return TSDB_CODE_SUCCESS;	2,001,391✔
150
151	_error:	×
152	if (pInfo != NULL) {	×
153	destroyEWindowOperatorInfo(pInfo);	×
154	}
155
156	destroyOperatorAndDownstreams(pOperator, &downstream, 1);	×
157	pTaskInfo->code = code;	×
158	return code;	×
159	}
160
161	void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo) {	2,002,855✔
162	if (pInfo == NULL \|\| pInfo->pRow == NULL \|\| pOperator == NULL) {	2,002,855✔
163	return;	1,168,342✔
164	}
165	SExprSupp* pSup = &pOperator->exprSupp;	834,513✔
166	for (int32_t j = 0; j < pSup->numOfExprs; ++j) {	3,224,061✔
167	pSup->pCtx[j].resultInfo = getResultEntryInfo(pInfo->pRow, j, pSup->rowEntryInfoOffset);	2,389,548✔
168	if (pSup->pCtx[j].fpSet.cleanup) {	2,389,548✔
169	pSup->pCtx[j].fpSet.cleanup(&pSup->pCtx[j]);	×
170	}
171	}
172	}
173
174	void destroyEWindowOperatorInfo(void* param) {	2,001,391✔
175	SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;	2,001,391✔
176	if (pInfo == NULL) {	2,001,391✔
177	return;	×
178	}
179
180	if (pInfo->pRow != NULL) {	2,001,391✔
181	taosMemoryFree(pInfo->pRow);	834,513✔
182	}
183
184	if (pInfo->pStartCondInfo != NULL) {	2,001,391✔
185	filterFreeInfo(pInfo->pStartCondInfo);	2,001,391✔
186	pInfo->pStartCondInfo = NULL;	2,001,391✔
187	}
188
189	if (pInfo->pEndCondInfo != NULL) {	2,001,391✔
190	filterFreeInfo(pInfo->pEndCondInfo);	2,001,391✔
191	pInfo->pEndCondInfo = NULL;	2,001,391✔
192	}
193
194	cleanupBasicInfo(&pInfo->binfo);	2,001,391✔
195	colDataDestroy(&pInfo->twAggSup.timeWindowData);	2,001,391✔
196
197	cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);	2,001,391✔
198	pInfo->pOperator = NULL;	2,001,391✔
199	cleanupAggSup(&pInfo->aggSup);	2,001,391✔
200	cleanupExprSupp(&pInfo->scalarSup);	2,001,391✔
201	taosMemoryFreeClear(param);	2,001,391✔
202	}
203
204	static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {	2,734,026✔
205	int32_t code = TSDB_CODE_SUCCESS;	2,734,026✔
206	int32_t lino = 0;	2,734,026✔
207	SEventWindowOperatorInfo* pInfo = pOperator->info;	2,734,026✔
208	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	2,734,026✔
209
210	SExprSupp* pSup = &pOperator->exprSupp;	2,734,026✔
211	int32_t order = pInfo->binfo.inputTsOrder;	2,734,026✔
212
213	SSDataBlock* pRes = pInfo->binfo.pRes;	2,734,026✔
214
215	blockDataCleanup(pRes);	2,734,026✔
216
217	SOperatorInfo* downstream = pOperator->pDownstream[0];	2,734,026✔
218	while (1) {	2,433,672✔
219	SSDataBlock* pBlock = NULL;	5,167,698✔
220	if (pInfo->pPreDataBlock == NULL) {	5,167,698✔
221	pBlock = getNextBlockFromDownstream(pOperator, 0);	4,930,203✔
222	} else {
223	pBlock = pInfo->pPreDataBlock;	237,495✔
224	pInfo->pPreDataBlock = NULL;	237,495✔
225	}
226
227	if (pBlock == NULL) {	5,167,698✔
228	break;	1,930,572✔
229	}
230
231	pRes->info.scanFlag = pBlock->info.scanFlag;	3,237,126✔
232	code = setInputDataBlock(pSup, pBlock, order, pBlock->info.scanFlag, true);	3,237,126✔
233	QUERY_CHECK_CODE(code, lino, _end);	3,237,126✔
234
235	code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);	3,237,126✔
236	QUERY_CHECK_CODE(code, lino, _end);	3,237,126✔
237
238	// there is an scalar expression that needs to be calculated right before apply the group aggregation.
239	if (pInfo->scalarSup.pExprInfo != NULL) {	3,237,126✔
240	code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,	1,235,279✔
241	pInfo->scalarSup.numOfExprs, NULL, GET_STM_RTINFO(pOperator->pTaskInfo));	1,235,279✔
242	QUERY_CHECK_CODE(code, lino, _end);	1,235,279✔
243	}
244
245	code = eventWindowAggImpl(pOperator, pInfo, pBlock);	2,783,685✔
246	QUERY_CHECK_CODE(code, lino, _end);	2,783,685✔
247
248	code = doFilter(pRes, pSup->pFilterInfo, NULL, NULL);	2,680,452✔
249	QUERY_CHECK_CODE(code, lino, _end);	2,680,452✔
250
251	if (pRes->info.rows >= pOperator->resultInfo.threshold \|\|	2,680,452✔
252	(pRes->info.id.groupId != pInfo->groupId && pRes->info.rows > 0)) {	2,644,724✔
253	(*ppRes) = pRes;	246,780✔
254	return code;	246,780✔
255	}
256	}
257
258	_end:	2,487,246✔
259	if (code != TSDB_CODE_SUCCESS) {	2,487,246✔
260	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	556,674✔
261	pTaskInfo->code = code;	556,674✔
262	T_LONG_JMP(pTaskInfo->env, code);	556,674✔
263	}
264	(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;	1,930,572✔
265	return code;	1,930,572✔
266	}
267
268	static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,	236,858,148✔
269	SExprSupp* pExprSup, SAggSupporter* pAggSup) {
270	if (*pResult == NULL) {	236,858,148✔
271	SResultRow* p = taosMemoryCalloc(1, pAggSup->resultRowSize);	835,977✔
272	if (!p) {	835,977✔
273	return terrno;	×
274	}
275	pResultRowInfo->cur = (SResultRowPosition){.pageId = p->pageId, .offset = p->offset};	835,977✔
276	*pResult = p;	835,977✔
277	}
278
279	(pResult)->win = win;	236,858,148✔
280
281	return setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	236,858,148✔
282	}
283
284	static int32_t doEventWindowAggImpl(SEventWindowOperatorInfo* pInfo, SExprSupp* pSup, int32_t startIndex,	236,858,148✔
285	int32_t endIndex, const SSDataBlock* pBlock, int64_t* tsList,
286	SExecTaskInfo* pTaskInfo) {
287	int32_t code = TSDB_CODE_SUCCESS;	236,858,148✔
288	SWindowRowsSup* pRowSup = &pInfo->winSup;	236,858,148✔
289
290	int32_t numOfOutput = pSup->numOfExprs;	236,858,148✔
291	int32_t numOfRows = endIndex - startIndex + 1;	236,858,148✔
292
293	doKeepTuple(pRowSup, tsList[endIndex], endIndex, pBlock->info.id.groupId);	236,858,148✔
294
295	code = setSingleOutputTupleBufv1(&pInfo->binfo.resultRowInfo, &pRowSup->win, &pInfo->pRow, pSup, &pInfo->aggSup);	236,858,148✔
296	if (code != TSDB_CODE_SUCCESS) { // null data, too many state code	236,858,148✔
297	qError("failed to set single output tuple buffer, code:%d", code);	×
298	return code;	×
299	}
300
301	updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, 0);	236,858,148✔
302	code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startIndex, numOfRows,	236,858,148✔
303	pBlock->info.rows, numOfOutput);	236,858,148✔
304	return code;	236,858,148✔
305	}
306
307	int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {	2,783,685✔
308	int32_t code = TSDB_CODE_SUCCESS;	2,783,685✔
309	int32_t lino = 0;	2,783,685✔
310	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	2,783,685✔
311	SExprSupp* pSup = &pOperator->exprSupp;	2,783,685✔
312	SSDataBlock* pRes = pInfo->binfo.pRes;	2,783,685✔
313	int64_t gid = pBlock->info.id.groupId;	2,783,685✔
314	SColumnInfoData ps = NULL, pe = NULL;	2,783,685✔
315	SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);	2,783,685✔
316	QUERY_CHECK_NULL(pColInfoData, code, lino, _return, terrno);	2,783,685✔
317	TSKEY* tsList = (TSKEY*)pColInfoData->pData;	2,783,685✔
318	SWindowRowsSup* pRowSup = &pInfo->winSup;	2,783,685✔
319	int32_t rowIndex = 0;	2,783,685✔
320	int64_t minWindowSize = getMinWindowSize(pOperator);	2,783,685✔
321
322	pRowSup->numOfRows = 0;	2,783,685✔
323	if (pInfo->groupId == 0) {	2,783,685✔
324	pInfo->groupId = gid;	2,280,197✔
325	} else if (pInfo->groupId != gid) {	503,488✔
326	// this is a new group, reset the info
327	pInfo->inWindow = false;	256,759✔
328	pInfo->groupId = gid;	256,759✔
329	pInfo->winSup.lastTs = INT64_MIN;	256,759✔
330	pInfo->pPreDataBlock = pBlock;	256,759✔
331	goto _return;	256,759✔
332	}
333	pRes->info.id.groupId = pInfo->groupId;	2,526,926✔
334
335	SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	2,526,926✔
336
337	code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &param1);	2,526,926✔
338	QUERY_CHECK_CODE(code, lino, _return);	2,526,926✔
339
340	int32_t status1 = 0;	2,526,926✔
341	code = filterExecute(pInfo->pStartCondInfo, pBlock, &ps, NULL, param1.numOfCols, &status1);	2,526,926✔
342	QUERY_CHECK_CODE(code, lino, _return);	2,526,926✔
343
344	SFilterColumnParam param2 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	2,526,926✔
345	code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &param2);	2,526,926✔
346	QUERY_CHECK_CODE(code, lino, _return);	2,526,926✔
347
348	int32_t status2 = 0;	2,526,926✔
349	code = filterExecute(pInfo->pEndCondInfo, pBlock, &pe, NULL, param2.numOfCols, &status2);	2,526,926✔
350	QUERY_CHECK_CODE(code, lino, _return);	2,526,926✔
351
352	for (int32_t i = 0; i < pBlock->info.rows; ++i) {	1,837,195,150✔
353	if (pBlock->info.scanFlag != PRE_SCAN) {	1,834,771,457✔
354	if (pInfo->winSup.lastTs == INT64_MIN) {	1,834,771,457✔
355	pInfo->winSup.lastTs = tsList[i];	1,182,938✔
356	} else {
357	if (tsList[i] == pInfo->winSup.lastTs) {	1,833,588,519✔
358	qError("duplicate timestamp found in event window operator, groupId: %" PRId64 ", timestamp: %" PRId64,	103,233✔
359	gid, tsList[i]);
360	code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;	103,233✔
361	QUERY_CHECK_CODE(code, lino, _return);	103,233✔
362	} else {
363	pInfo->winSup.lastTs = tsList[i];	1,833,485,286✔
364	}
365	}
366	}
367	}
368	int32_t startIndex = pInfo->inWindow ? 0 : -1;	2,423,693✔
369	while (rowIndex < pBlock->info.rows) {	475,283,551✔
370	if (pInfo->inWindow) { // let's find the first end value	473,308,337✔
371	for (rowIndex = startIndex; rowIndex < pBlock->info.rows; ++rowIndex) {	1,352,217,881✔
372	if (((bool*)pe->pData)[rowIndex]) {	1,350,970,687✔
373	break;	235,610,954✔
374	}
375	}
376
377	if (rowIndex < pBlock->info.rows) {	236,858,148✔
378	code = doEventWindowAggImpl(pInfo, pSup, startIndex, rowIndex, pBlock, tsList, pTaskInfo);	235,610,954✔
379	QUERY_CHECK_CODE(code, lino, _return);	235,610,954✔
380	doUpdateNumOfRows(pSup->pCtx, pInfo->pRow, pSup->numOfExprs, pSup->rowEntryInfoOffset);	235,610,954✔
381
382	int64_t delta = pRowSup->win.ekey - pRowSup->win.skey;	235,610,954✔
383	if(pInfo->binfo.inputTsOrder == ORDER_DESC) {	235,610,954✔
384	delta = -delta;	1,512✔
385	}
386	if (delta < minWindowSize) {	235,610,954✔
387	qDebug("skip small window, groupId: %" PRId64 ", windowSize: %" PRId64 ", minWindowSize: %" PRId64,	1,860✔
388	pInfo->groupId, delta, minWindowSize);
389	} else {
390	// check buffer size
391	if (pRes->info.rows + pInfo->pRow->numOfRows >= pRes->info.capacity) {	235,609,094✔
392	int32_t newSize = pRes->info.rows + pInfo->pRow->numOfRows;	666,432✔
393	code = blockDataEnsureCapacity(pRes, newSize);	666,432✔
394	QUERY_CHECK_CODE(code, lino, _return);	666,432✔
395	}
396
397	code = copyResultrowToDataBlock(pSup->pExprInfo, pSup->numOfExprs, pInfo->pRow, pSup->pCtx, pRes,	235,609,094✔
398	pSup->rowEntryInfoOffset, pTaskInfo);	235,609,094✔
399	QUERY_CHECK_CODE(code, lino, _return);	235,609,094✔
400
401	pRes->info.rows += pInfo->pRow->numOfRows;	235,609,094✔
402	}
403	pInfo->pRow->numOfRows = 0;	235,610,954✔
404
405	pInfo->inWindow = false;	235,610,954✔
406	rowIndex += 1;	235,610,954✔
407	} else {
408	code = doEventWindowAggImpl(pInfo, pSup, startIndex, pBlock->info.rows - 1, pBlock, tsList, pTaskInfo);	1,247,194✔
409	QUERY_CHECK_CODE(code, lino, _return);	1,247,194✔
410	}
411	} else { // find the first start value that is fulfill for the start condition
412	for (; rowIndex < pBlock->info.rows; ++rowIndex) {	720,044,493✔
413	if (((bool*)ps->pData)[rowIndex]) {	719,596,014✔
414	doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);	236,001,710✔
415	pInfo->inWindow = true;	236,001,710✔
416	startIndex = rowIndex;	236,001,710✔
417	if (pInfo->pRow != NULL) {	236,001,710✔
418	clearResultRowInitFlag(pSup->pCtx, pSup->numOfExprs);	235,165,733✔
419	}
420	break;	236,001,710✔
421	}
422	}
423
424	if (pInfo->inWindow) {	236,450,565✔
425	continue; // try to find the end position	236,001,710✔
426	} else {
427	break; // no valid start position, quit	448,855✔
428	}
429	}
430	}
431
432	_return:	2,783,685✔
433
434	if (code != TSDB_CODE_SUCCESS) {	2,783,685✔
435	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	103,233✔
436	}
437	colDataDestroy(ps);	2,783,685✔
438	taosMemoryFree(ps);	2,783,685✔
439	colDataDestroy(pe);	2,783,685✔
440	taosMemoryFree(pe);	2,783,685✔
441
442	return code;	2,783,685✔
443	}

taosdata / TDengine / #4932

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