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

32

static int32_t resetEventWindowOperState(SOperatorInfo* pOper) {

1,440✔

33

  SEventWindowOperatorInfo* pEvent = pOper->info;

1,440✔

34

  SExecTaskInfo*           pTaskInfo = pOper->pTaskInfo;

1,440✔

35

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

1,440✔

36

  pOper->status = OP_NOT_OPENED;

1,440✔

38

  resetBasicOperatorState(&pEvent->binfo);

1,440✔

39

  taosMemoryFreeClear(pEvent->pRow);

1,440✔

41

  pEvent->groupId = 0;

1,440✔

42

  pEvent->pPreDataBlock = NULL;

1,440✔

43

  pEvent->inWindow = false;

1,440✔

44

  pEvent->winSup.lastTs = INT64_MIN;

1,440✔

46

  colDataDestroy(&pEvent->twAggSup.timeWindowData);

1,440✔

47

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

1,440✔

48

  cleanupResultInfoInEventWindow(pOper, pEvent);

1,440✔

50

  if (code == 0) {

1,440✔

51

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

2,880✔

52

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

1,440✔

55

  if (code == 0) {

1,440✔

56

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

1,440✔

59

  return code;

1,440✔

62

int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,

208,317✔

64

  QRY_PARAM_CHECK(pOptrInfo);

208,317✔

66

  int32_t                   code = TSDB_CODE_SUCCESS;

208,695✔

67

  int32_t                   lino = 0;

208,695✔

68

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

208,695✔

69

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

208,317✔

70

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

208,317✔

75

  pOperator->pPhyNode = physiNode;

208,317✔

76

  pOperator->exprSupp.hasWindowOrGroup = true;

208,317✔

77

  pOperator->exprSupp.hasWindow = true;

208,317✔

79

  SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;

208,695✔

81

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

208,695✔

82

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

208,317✔

83

                            pTaskInfo->pStreamRuntimeInfo);

208,695✔

84

  QUERY_CHECK_CODE(code, lino, _error);

208,695✔

86

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

208,695✔

87

                            pTaskInfo->pStreamRuntimeInfo);

208,695✔

88

  QUERY_CHECK_CODE(code, lino, _error);

208,695✔

90

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

208,695✔

91

    int32_t    numOfScalarExpr = 0;

46,861✔

92

    SExprInfo* pScalarExprInfo = NULL;

46,861✔

94

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

46,861✔

95

    QUERY_CHECK_CODE(code, lino, _error);

46,861✔

96

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

46,861✔

97

    QUERY_CHECK_CODE(code, lino, _error);

46,861✔

100

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

208,695✔

101

                            pTaskInfo->pStreamRuntimeInfo);

208,695✔

102

  QUERY_CHECK_CODE(code, lino, _error);

208,695✔

104

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

208,695✔

106

  int32_t    num = 0;

208,695✔

107

  SExprInfo* pExprInfo = NULL;

208,695✔

108

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

208,695✔

109

  QUERY_CHECK_CODE(code, lino, _error);

208,695✔

111

  initResultSizeInfo(&pOperator->resultInfo, 4096);

208,695✔

113

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

417,390✔

114

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

208,695✔

115

  QUERY_CHECK_CODE(code, lino, _error);

208,317✔

117

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

208,317✔

118

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

208,695✔

119

  initBasicInfo(&pInfo->binfo, pResBlock);

208,695✔

121

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

208,695✔

122

  QUERY_CHECK_CODE(code, lino, _error);

208,695✔

124

  initResultRowInfo(&pInfo->binfo.resultRowInfo);

208,695✔

125

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

208,695✔

126

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

208,695✔

127

  pInfo->winSup.lastTs = INT64_MIN;

208,695✔

129

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

208,695✔

130

  QUERY_CHECK_CODE(code, lino, _error);

208,695✔

132

  pInfo->tsSlotId = tsSlotId;

208,695✔

133

  pInfo->pPreDataBlock = NULL;

208,695✔

134

  pInfo->pOperator = pOperator;

208,695✔

135

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

208,695✔

136

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

208,695✔

137

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

208,695✔

139

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

208,695✔

141

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

208,695✔

144

  setOperatorResetStateFn(pOperator, resetEventWindowOperState);

208,695✔

145

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

208,695✔

146

  if (code != TSDB_CODE_SUCCESS) {

208,695✔

150

  *pOptrInfo = pOperator;

208,695✔

151

  return TSDB_CODE_SUCCESS;

208,695✔

163

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

210,135✔

164

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

210,135✔

165

    return;

63,977✔

167

  SExprSupp*       pSup = &pOperator->exprSupp;

146,158✔

168

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

146,158✔

176

void destroyEWindowOperatorInfo(void* param) {

208,695✔

177

  SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;

208,695✔

178

  if (pInfo == NULL) {

208,695✔

184

  cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);

208,695✔

186

  if (pInfo->pRow != NULL) {

208,695✔

187

    taosMemoryFree(pInfo->pRow);

146,158✔

188

    pInfo->pRow = NULL;

146,158✔

191

  if (pInfo->pStartCondInfo != NULL) {

208,695✔

192

    filterFreeInfo(pInfo->pStartCondInfo);

208,695✔

193

    pInfo->pStartCondInfo = NULL;

208,695✔

196

  if (pInfo->pEndCondInfo != NULL) {

208,695✔

197

    filterFreeInfo(pInfo->pEndCondInfo);

208,317✔

198

    pInfo->pEndCondInfo = NULL;

208,695✔

201

  cleanupBasicInfo(&pInfo->binfo);

209,073✔

202

  colDataDestroy(&pInfo->twAggSup.timeWindowData);

208,317✔

203

  pInfo->pOperator = NULL;

208,695✔

204

  cleanupAggSup(&pInfo->aggSup);

208,695✔

205

  cleanupExprSupp(&pInfo->scalarSup);

208,695✔

206

  taosMemoryFreeClear(param);

208,695✔

209

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

482,069✔

210

  int32_t                   code = TSDB_CODE_SUCCESS;

482,069✔

211

  int32_t                   lino = 0;

482,069✔

212

  SEventWindowOperatorInfo* pInfo = pOperator->info;

482,069✔

213

  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;

482,069✔

215

  SExprSupp* pSup = &pOperator->exprSupp;

482,069✔

216

  int32_t    order = pInfo->binfo.inputTsOrder;

482,069✔

218

  SSDataBlock* pRes = pInfo->binfo.pRes;

482,069✔

220

  blockDataCleanup(pRes);

482,069✔

222

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

482,069✔

223

  while (1) {

1,190,251✔

224

    SSDataBlock* pBlock = NULL;

1,672,320✔

225

    if (pInfo->pPreDataBlock == NULL) {

1,672,320✔

226

      pBlock = getNextBlockFromDownstream(pOperator, 0);

1,407,170✔

228

      pBlock = pInfo->pPreDataBlock;

265,150✔

229

      pInfo->pPreDataBlock = NULL;

265,150✔

232

    if (pBlock == NULL) {

1,672,320✔

233

      break;

194,374✔

236

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

1,477,946✔

237

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

1,477,946✔

238

    QUERY_CHECK_CODE(code, lino, _end);

1,477,946✔

240

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

1,477,946✔

241

    QUERY_CHECK_CODE(code, lino, _end);

1,477,946✔

244

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

1,477,946✔

245

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

41,377✔

246

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

41,377✔

247

      QUERY_CHECK_CODE(code, lino, _end);

41,377✔

250

    code = eventWindowAggImpl(pOperator, pInfo, pBlock);

1,463,906✔

251

    QUERY_CHECK_CODE(code, lino, _end);

1,463,906✔

253

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

1,453,737✔

254

    QUERY_CHECK_CODE(code, lino, _end);

1,453,737✔

256

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

1,453,737✔

257

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

1,429,729✔

258

      (*ppRes) = pRes;

263,486✔

259

      return code;

263,486✔

263

_end:

218,583✔

264

  if (code != TSDB_CODE_SUCCESS) {

218,583✔

265

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

24,209✔

266

    pTaskInfo->code = code;

24,209✔

267

    T_LONG_JMP(pTaskInfo->env, code);

24,209✔

269

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

194,374✔

270

  return code;

194,374✔

273

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

159,025,111✔

275

  if (*pResult == NULL) {

159,025,111✔

276

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

147,598✔

277

    if (!p) {

147,598✔

280

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

147,598✔

281

    *pResult = p;

147,598✔

284

  (*pResult)->win = *win;

159,025,111✔

286

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

159,025,111✔

289

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

159,025,111✔

292

  int32_t code = TSDB_CODE_SUCCESS;

159,025,111✔

293

  SWindowRowsSup* pRowSup = &pInfo->winSup;

159,025,111✔

295

  int32_t numOfOutput = pSup->numOfExprs;

159,025,111✔

296

  int32_t numOfRows = endIndex - startIndex + 1;

159,025,111✔

298

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

159,025,111✔

300

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

159,025,111✔

301

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

159,025,111✔

306

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

159,025,111✔

307

  pInfo->pRow->nOrigRows += numOfRows;

159,025,111✔

308

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

159,025,111✔

309

                                         pBlock->info.rows, numOfOutput);

159,025,111✔

310

  return code;

159,025,111✔

313

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

1,463,906✔

314

  int32_t          code = TSDB_CODE_SUCCESS;

1,463,906✔

315

  int32_t          lino = 0;

1,463,906✔

316

  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;

1,463,906✔

317

  SExprSupp*       pSup = &pOperator->exprSupp;

1,463,906✔

318

  SSDataBlock*     pRes = pInfo->binfo.pRes;

1,463,906✔

319

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

1,463,906✔

320

  SColumnInfoData *ps = NULL, *pe = NULL;

1,463,906✔

321

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

1,463,906✔

322

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

1,463,906✔

323

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

1,463,906✔

324

  SWindowRowsSup*  pRowSup = &pInfo->winSup;

1,463,906✔

325

  int32_t          rowIndex = 0;

1,463,906✔

326

  STrueForInfo*    pTrueForInfo = getTrueForInfo(pOperator);

1,463,906✔

328

  pRowSup->numOfRows = 0;

1,463,906✔

329

  if (pInfo->groupId == 0) {

1,463,906✔

330

    pInfo->groupId = gid;

905,482✔

331

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

558,424✔

333

    pInfo->inWindow = false;

284,202✔

334

    pInfo->groupId = gid;

284,202✔

335

    pInfo->winSup.lastTs = INT64_MIN;

284,202✔

336

    pInfo->pPreDataBlock = pBlock;

284,202✔

337

    goto _return;

284,202✔

339

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

1,179,704✔

341

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

1,179,704✔

343

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

1,179,704✔

344

  QUERY_CHECK_CODE(code, lino, _return);

1,179,704✔

346

  int32_t status1 = 0;

1,179,704✔

347

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

1,179,704✔

348

  QUERY_CHECK_CODE(code, lino, _return);

1,179,704✔

350

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

1,179,704✔

351

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

1,179,704✔

352

  QUERY_CHECK_CODE(code, lino, _return);

1,179,704✔

354

  int32_t status2 = 0;

1,179,704✔

355

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

1,179,704✔

356

  QUERY_CHECK_CODE(code, lino, _return);

1,179,704✔

358

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

1,239,262,508✔

359

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

1,238,092,973✔

360

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

1,238,095,696✔

361

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

428,637✔

363

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

1,237,667,448✔

364

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

10,169✔

366

          code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;

8,224✔

367

          QUERY_CHECK_CODE(code, lino, _return);

8,224✔

369

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

1,237,656,112✔

374

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

1,169,535✔

375

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

318,739,680✔

376

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

317,896,723✔

377

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

903,011,950✔

378

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

902,319,973✔

379

          break;

158,333,134✔

383

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

159,025,111✔

384

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

158,333,134✔

385

        QUERY_CHECK_CODE(code, lino, _return);

158,333,134✔

386

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

158,333,134✔

388

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

158,333,134✔

393

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

158,333,134✔

394

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

658,656✔

395

            code = blockDataEnsureCapacity(pRes, newSize);

658,656✔

396

            QUERY_CHECK_CODE(code, lino, _return);

658,656✔

399

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

158,333,134✔

400

                                          pSup->rowEntryInfoOffset, pTaskInfo);

158,333,134✔

401

          QUERY_CHECK_CODE(code, lino, _return);

158,333,134✔

403

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

158,333,134✔

405

        pInfo->pRow->numOfRows = 0;

158,333,134✔

406

        pInfo->pRow->nOrigRows = 0;

158,333,134✔

408

        pInfo->inWindow = false;

158,333,134✔

409

        rowIndex += 1;

158,333,134✔

411

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

691,977✔

412

        QUERY_CHECK_CODE(code, lino, _return);

691,977✔

415

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

494,712,188✔

416

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

494,385,610✔

417

          doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);

158,545,034✔

418

          pInfo->inWindow = true;

158,545,034✔

419

          startIndex = rowIndex;

158,545,034✔

420

          if (pInfo->pRow != NULL) {

158,545,034✔

421

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

158,397,436✔

423

          break;

158,545,034✔

427

      if (pInfo->inWindow) {

158,871,612✔

428

        continue;  // try to find the end position

158,545,034✔

430

        break;  // no valid start position, quit

326,578✔

435

_return:

1,463,906✔

437

  if (code != TSDB_CODE_SUCCESS) {

1,463,906✔

438

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

10,169✔

440

  colDataDestroy(ps);

1,463,906✔

441

  taosMemoryFree(ps);

1,463,906✔

442

  colDataDestroy(pe);

1,463,906✔

443

  taosMemoryFree(pe);

1,463,906✔

445

  return code;

1,463,906✔

taosdata / TDengine / #4944

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

taosdata / TDengine / #4944

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