• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #3535

23 Nov 2024 02:07AM UTC coverage: 60.85% (+0.03%) from 60.825%
#3535

push

travis-ci

web-flow
Merge pull request #28893 from taosdata/doc/internal

refact: rename taos lib name

120252 of 252737 branches covered (47.58%)

Branch coverage included in aggregate %.

201187 of 275508 relevant lines covered (73.02%)

15886166.19 hits per line

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

73.1
/source/libs/executor/src/projectoperator.c
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 "functionMgt.h"
19
#include "operator.h"
20
#include "querytask.h"
21
#include "tdatablock.h"
22

23
typedef struct SProjectOperatorInfo {
24
  SOptrBasicInfo binfo;
25
  SAggSupporter  aggSup;
26
  SArray*        pPseudoColInfo;
27
  SLimitInfo     limitInfo;
28
  bool           mergeDataBlocks;
29
  SSDataBlock*   pFinalRes;
30
  bool           inputIgnoreGroup;
31
  bool           outputIgnoreGroup;
32
} SProjectOperatorInfo;
33

34
typedef struct SIndefOperatorInfo {
35
  SOptrBasicInfo binfo;
36
  SAggSupporter  aggSup;
37
  SArray*        pPseudoColInfo;
38
  SExprSupp      scalarSup;
39
  uint64_t       groupId;
40
  SSDataBlock*   pNextGroupRes;
41
} SIndefOperatorInfo;
42

43
static int32_t      doGenerateSourceData(SOperatorInfo* pOperator);
44
static SSDataBlock* doProjectOperation1(SOperatorInfo* pOperator);
45
static int32_t      doProjectOperation(SOperatorInfo* pOperator, SSDataBlock** pResBlock);
46
static SSDataBlock* doApplyIndefinitFunction1(SOperatorInfo* pOperator);
47
static int32_t      doApplyIndefinitFunction(SOperatorInfo* pOperator, SSDataBlock** pResBlock);
48
static int32_t      setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols, SArray** pResList);
49
static int32_t      setFunctionResultOutput(SOperatorInfo* pOperator, SOptrBasicInfo* pInfo, SAggSupporter* pSup,
50
                                            int32_t stage, int32_t numOfExprs);
51

52
static void destroyProjectOperatorInfo(void* param) {
1,697,473✔
53
  if (NULL == param) {
1,697,473!
54
    return;
×
55
  }
56

57
  SProjectOperatorInfo* pInfo = (SProjectOperatorInfo*)param;
1,697,473✔
58
  cleanupBasicInfo(&pInfo->binfo);
1,697,473✔
59
  cleanupAggSup(&pInfo->aggSup);
1,697,510✔
60
  taosArrayDestroy(pInfo->pPseudoColInfo);
1,697,513✔
61

62
  blockDataDestroy(pInfo->pFinalRes);
1,697,499✔
63
  taosMemoryFreeClear(param);
1,697,524✔
64
}
65

66
static void destroyIndefinitOperatorInfo(void* param) {
122,343✔
67
  SIndefOperatorInfo* pInfo = (SIndefOperatorInfo*)param;
122,343✔
68
  if (pInfo == NULL) {
122,343!
69
    return;
×
70
  }
71

72
  cleanupBasicInfo(&pInfo->binfo);
122,343✔
73
  taosArrayDestroy(pInfo->pPseudoColInfo);
122,344✔
74
  cleanupAggSup(&pInfo->aggSup);
122,344✔
75
  cleanupExprSupp(&pInfo->scalarSup);
122,344✔
76

77
  taosMemoryFreeClear(param);
122,344!
78
}
79

80
void streamOperatorReleaseState(SOperatorInfo* pOperator) {
1,178✔
81
  SOperatorInfo* downstream = pOperator->pDownstream[0];
1,178✔
82
  if (downstream->fpSet.releaseStreamStateFn) {
1,178!
83
    downstream->fpSet.releaseStreamStateFn(downstream);
1,180✔
84
  }
85
}
1,178✔
86

87
void streamOperatorReloadState(SOperatorInfo* pOperator) {
1,181✔
88
  SOperatorInfo* downstream = pOperator->pDownstream[0];
1,181✔
89
  if (downstream->fpSet.reloadStreamStateFn) {
1,181!
90
    downstream->fpSet.reloadStreamStateFn(downstream);
1,181✔
91
  }
92
}
1,181✔
93

94
int32_t createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo,
1,696,603✔
95
                                  SOperatorInfo** pOptrInfo) {
96
  QRY_PARAM_CHECK(pOptrInfo);
1,696,603!
97

98
  int32_t code = TSDB_CODE_SUCCESS;
1,696,603✔
99
  SProjectOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SProjectOperatorInfo));
1,696,603✔
100
  SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1,697,036✔
101
  if (pInfo == NULL || pOperator == NULL) {
1,697,125!
102
    code = terrno;
22✔
103
    goto _error;
×
104
  }
105

106
  pOperator->exprSupp.hasWindowOrGroup = false;
1,697,103✔
107
  pOperator->pTaskInfo = pTaskInfo;
1,697,103✔
108

109
  int32_t    lino = 0;
1,697,103✔
110

111
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pProjPhyNode->node.pOutputDataBlockDesc);
1,697,103✔
112
  TSDB_CHECK_NULL(pResBlock, code, lino, _error, terrno);
1,697,493!
113

114
  initLimitInfo(pProjPhyNode->node.pLimit, pProjPhyNode->node.pSlimit, &pInfo->limitInfo);
1,697,493✔
115

116
  pInfo->binfo.pRes = pResBlock;
1,697,477✔
117
  pInfo->pFinalRes = NULL;
1,697,477✔
118

119
  code = createOneDataBlock(pResBlock, false, &pInfo->pFinalRes);
1,697,477✔
120
  TSDB_CHECK_CODE(code, lino, _error);
1,697,249!
121

122
  pInfo->binfo.inputTsOrder = pProjPhyNode->node.inputTsOrder;
1,697,249✔
123
  pInfo->binfo.outputTsOrder = pProjPhyNode->node.outputTsOrder;
1,697,249✔
124
  pInfo->inputIgnoreGroup = pProjPhyNode->inputIgnoreGroup;
1,697,249✔
125
  pInfo->outputIgnoreGroup = pProjPhyNode->ignoreGroupId;
1,697,249✔
126

127
  if (pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM || pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
1,697,249✔
128
    pInfo->mergeDataBlocks = false;
4,855✔
129
  } else {
130
    if (!pProjPhyNode->ignoreGroupId) {
1,692,394✔
131
      pInfo->mergeDataBlocks = false;
32,793✔
132
    } else {
133
      pInfo->mergeDataBlocks = pProjPhyNode->mergeDataBlock;
1,659,601✔
134
    }
135
  }
136

137
  int32_t numOfRows = 4096;
1,697,249✔
138
  size_t  keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
1,697,249✔
139

140
  // Make sure the size of SSDataBlock will never exceed the size of 2MB.
141
  int32_t TWOMB = 2 * 1024 * 1024;
1,697,249✔
142
  if (numOfRows * pResBlock->info.rowSize > TWOMB) {
1,697,249✔
143
    numOfRows = TWOMB / pResBlock->info.rowSize;
260,790✔
144
  }
145

146
  initResultSizeInfo(&pOperator->resultInfo, numOfRows);
1,697,249✔
147
  
148
  int32_t    numOfCols = 0;
1,697,224✔
149
  SExprInfo* pExprInfo = NULL;
1,697,224✔
150
  code = createExprInfo(pProjPhyNode->pProjections, NULL, &pExprInfo, &numOfCols);
1,697,224✔
151
  TSDB_CHECK_CODE(code, lino, _error);
1,697,384!
152
  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str,
1,697,384✔
153
                    pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);
1,697,384✔
154
  TSDB_CHECK_CODE(code, lino, _error);
1,697,176!
155

156
  initBasicInfo(&pInfo->binfo, pResBlock);
1,697,176✔
157
  code = setFunctionResultOutput(pOperator, &pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfCols);
1,697,098✔
158
  TSDB_CHECK_CODE(code, lino, _error);
1,697,150!
159

160
  code = filterInitFromNode((SNode*)pProjPhyNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
1,697,150✔
161
  TSDB_CHECK_CODE(code, lino, _error);
1,696,728!
162

163
  code = setRowTsColumnOutputInfo(pOperator->exprSupp.pCtx, numOfCols, &pInfo->pPseudoColInfo);
1,696,728✔
164
  TSDB_CHECK_CODE(code, lino, _error);
1,697,327!
165

166
  setOperatorInfo(pOperator, "ProjectOperator", QUERY_NODE_PHYSICAL_PLAN_PROJECT, false, OP_NOT_OPENED, pInfo,
1,697,327✔
167
                  pTaskInfo);
168
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doProjectOperation, NULL, destroyProjectOperatorInfo,
1,696,518✔
169
                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
170
  setOperatorStreamStateFn(pOperator, streamOperatorReleaseState, streamOperatorReloadState);
1,696,242✔
171

172
  if (NULL != downstream) {
1,696,745✔
173
    code = appendDownstream(pOperator, &downstream, 1);
1,695,249✔
174
    if (code != TSDB_CODE_SUCCESS) {
1,695,955!
175
      goto _error;
×
176
    }
177
  }
178

179
  *pOptrInfo = pOperator;
1,697,451✔
180
  return TSDB_CODE_SUCCESS;
1,697,451✔
181

182
_error:
×
183
  if (pInfo != NULL) destroyProjectOperatorInfo(pInfo);
×
184
  destroyOperatorAndDownstreams(pOperator, &downstream, 1);
×
185
  pTaskInfo->code = code;
×
186
  return code;
×
187
}
188

189
static int32_t discardGroupDataBlock(SSDataBlock* pBlock, SLimitInfo* pLimitInfo) {
3,668,902✔
190
  if (pLimitInfo->remainGroupOffset > 0) {
3,668,902✔
191
    // it is the first group
192
    if (pLimitInfo->currentGroupId == 0 || pLimitInfo->currentGroupId == pBlock->info.id.groupId) {
3,783✔
193
      pLimitInfo->currentGroupId = pBlock->info.id.groupId;
2,090✔
194
      return PROJECT_RETRIEVE_CONTINUE;
2,090✔
195
    } else if (pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
1,693!
196
      // now it is the data from a new group
197
      pLimitInfo->remainGroupOffset -= 1;
1,693✔
198
      pLimitInfo->currentGroupId = pBlock->info.id.groupId;
1,693✔
199

200
      // ignore data block in current group
201
      if (pLimitInfo->remainGroupOffset > 0) {
1,693✔
202
        return PROJECT_RETRIEVE_CONTINUE;
1,473✔
203
      }
204

205
      pLimitInfo->currentGroupId = 0;
220✔
206
    }
207
  }
208

209
  return PROJECT_RETRIEVE_DONE;
3,665,339✔
210
}
211

212
static int32_t setInfoForNewGroup(SSDataBlock* pBlock, SLimitInfo* pLimitInfo, SOperatorInfo* pOperator) {
3,665,279✔
213
  // remainGroupOffset == 0
214
  // here check for a new group data, we need to handle the data of the previous group.
215
  if (!(pLimitInfo->remainGroupOffset == 0 || pLimitInfo->remainGroupOffset == -1)) {
3,665,279!
216
    qError("project failed at: %s:%d", __func__, __LINE__);
×
217
    return TSDB_CODE_INVALID_PARA;
×
218
  }
219

220
  bool newGroup = false;
3,665,279✔
221
  if (0 == pBlock->info.id.groupId) {
3,665,279✔
222
    pLimitInfo->numOfOutputGroups = 1;
3,424,683✔
223
  } else if (pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
240,596✔
224
    pLimitInfo->currentGroupId = pBlock->info.id.groupId;
220,602✔
225
    pLimitInfo->numOfOutputGroups += 1;
220,602✔
226
    newGroup = true;
220,602✔
227
  } else {
228
    return PROJECT_RETRIEVE_CONTINUE;
19,994✔
229
  }
230

231
  if ((pLimitInfo->slimit.limit >= 0) && (pLimitInfo->slimit.limit < pLimitInfo->numOfOutputGroups)) {
3,645,285✔
232
    setOperatorCompleted(pOperator);
413✔
233
    return PROJECT_RETRIEVE_DONE;
413✔
234
  }
235

236
  // reset the value for a new group data
237
  // existing rows that belongs to previous group.
238
  if (newGroup) {
3,644,872✔
239
    resetLimitInfoForNextGroup(pLimitInfo);
220,191✔
240
  }
241

242
  return PROJECT_RETRIEVE_CONTINUE;
3,645,002✔
243
}
244

245
// todo refactor
246
static int32_t doIngroupLimitOffset(SLimitInfo* pLimitInfo, uint64_t groupId, SSDataBlock* pBlock,
3,666,634✔
247
                                    SOperatorInfo* pOperator) {
248
  // set current group id
249
  pLimitInfo->currentGroupId = groupId;
3,666,634✔
250
  bool limitReached = applyLimitOffset(pLimitInfo, pBlock, pOperator->pTaskInfo);
3,666,634✔
251
  if (pBlock->info.rows == 0 && 0 != pLimitInfo->limit.limit) {
3,666,732✔
252
    return PROJECT_RETRIEVE_CONTINUE;
17,065✔
253
  } else {
254
    if (limitReached && (pLimitInfo->slimit.limit >= 0 && pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
3,649,667✔
255
      setOperatorCompleted(pOperator);
116✔
256
    } else if (limitReached && groupId == 0) {
3,649,551✔
257
      setOperatorCompleted(pOperator);
109,540✔
258
    }
259
  }
260

261
  return PROJECT_RETRIEVE_DONE;
3,649,617✔
262
}
263

264
int32_t doProjectOperation(SOperatorInfo* pOperator, SSDataBlock** pResBlock) {
4,121,150✔
265
  QRY_PARAM_CHECK(pResBlock);
4,121,150!
266

267
  SProjectOperatorInfo* pProjectInfo = pOperator->info;
4,121,150✔
268
  SOptrBasicInfo*       pInfo = &pProjectInfo->binfo;
4,121,150✔
269
  SExprSupp*            pSup = &pOperator->exprSupp;
4,121,150✔
270
  SSDataBlock*          pRes = pInfo->pRes;
4,121,150✔
271
  SSDataBlock*          pFinalRes = pProjectInfo->pFinalRes;
4,121,150✔
272
  int32_t               code = 0;
4,121,150✔
273
  int32_t               lino = 0;
4,121,150✔
274
  int64_t               st = 0;
4,121,150✔
275
  int32_t               order = pInfo->inputTsOrder;
4,121,150✔
276
  int32_t               scanFlag = 0;
4,121,150✔
277

278
  blockDataCleanup(pFinalRes);
4,121,150✔
279
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
4,121,465✔
280

281
  if (pOperator->status == OP_EXEC_DONE) {
4,121,465✔
282
    return code;
973,466✔
283
  }
284

285
  if (pOperator->cost.openCost == 0) {
3,147,999✔
286
    st = taosGetTimestampUs();
1,697,854✔
287
  }
288

289
  SOperatorInfo* downstream = pOperator->numOfDownstream > 0 ? pOperator->pDownstream[0] : NULL;
3,148,661✔
290
  SLimitInfo*    pLimitInfo = &pProjectInfo->limitInfo;
3,148,661✔
291

292
  if (downstream == NULL) {
3,148,661✔
293
    code = doGenerateSourceData(pOperator);
1,591✔
294
    QUERY_CHECK_CODE(code, lino, _end);
1,591!
295

296
    if (pProjectInfo->outputIgnoreGroup) {
1,591!
297
      pRes->info.id.groupId = 0;
1,591✔
298
    }
299

300
    *pResBlock = (pRes->info.rows > 0)? pRes:NULL;
1,591!
301
    return code;
1,591✔
302
  }
303

304
  while (1) {
2,176,817✔
305
    while (1) {
20,628✔
306
      blockDataCleanup(pRes);
5,344,515✔
307

308
      // The downstream exec may change the value of the newgroup, so use a local variable instead.
309
      SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);
5,343,765✔
310
      if (pBlock == NULL) {
5,344,409✔
311
        qDebug("set op close, exec %d, status %d rows %" PRId64 , pTaskInfo->execModel, pOperator->status, pFinalRes->info.rows);
1,669,169✔
312
        setOperatorCompleted(pOperator);
1,669,170✔
313
        break;
1,669,241✔
314
      }
315
//      if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
316
//        qDebug("set status recv");
317
//        pOperator->status = OP_EXEC_RECV;
318
//      }
319

320
      // for stream interval
321
      if (pBlock->info.type == STREAM_RETRIEVE || pBlock->info.type == STREAM_DELETE_RESULT ||
3,675,240✔
322
          pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_CREATE_CHILD_TABLE ||
3,673,803!
323
          pBlock->info.type == STREAM_CHECKPOINT) {
3,669,833✔
324

325
        *pResBlock = pBlock;
6,073✔
326
        return code;
6,073✔
327
      }
328

329
      if (pProjectInfo->inputIgnoreGroup) {
3,669,167✔
330
        pBlock->info.id.groupId = 0;
469,587✔
331
      }
332

333
      int32_t status = discardGroupDataBlock(pBlock, pLimitInfo);
3,669,167✔
334
      if (status == PROJECT_RETRIEVE_CONTINUE) {
3,669,019✔
335
        continue;
3,563✔
336
      }
337

338
      (void) setInfoForNewGroup(pBlock, pLimitInfo, pOperator);
3,665,456✔
339
      if (pOperator->status == OP_EXEC_DONE) {
3,665,515✔
340
        break;
413✔
341
      }
342

343
      if (pProjectInfo->mergeDataBlocks) {
3,665,102✔
344
        pFinalRes->info.scanFlag = scanFlag = pBlock->info.scanFlag;
2,353,298✔
345
      } else {
346
        pRes->info.scanFlag = scanFlag = pBlock->info.scanFlag;
1,311,804✔
347
      }
348

349
      code = setInputDataBlock(pSup, pBlock, order, scanFlag, false);
3,665,102✔
350
      QUERY_CHECK_CODE(code, lino, _end);
3,665,498!
351

352
      code = blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
3,665,498✔
353
      QUERY_CHECK_CODE(code, lino, _end);
3,665,630!
354

355
      code = projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs,
3,665,630✔
356
                                   pProjectInfo->pPseudoColInfo);
357
      QUERY_CHECK_CODE(code, lino, _end);
3,665,172✔
358

359
      status = doIngroupLimitOffset(pLimitInfo, pBlock->info.id.groupId, pInfo->pRes, pOperator);
3,665,067✔
360
      if (status == PROJECT_RETRIEVE_CONTINUE) {
3,665,268✔
361
        continue;
17,065✔
362
      }
363

364
      break;
3,648,203✔
365
    }
366

367
    if (pProjectInfo->mergeDataBlocks) {
5,317,857✔
368
      if (pRes->info.rows > 0) {
3,324,171✔
369
        pFinalRes->info.id.groupId = 0;  // clear groupId
2,335,732✔
370
        pFinalRes->info.version = pRes->info.version;
2,335,732✔
371

372
        // continue merge data, ignore the group id
373
        code = blockDataMerge(pFinalRes, pRes);
2,335,732✔
374
        QUERY_CHECK_CODE(code, lino, _end);
2,336,206!
375

376
        if (pFinalRes->info.rows + pRes->info.rows <= pOperator->resultInfo.threshold && (pOperator->status != OP_EXEC_DONE)) {
2,336,206✔
377
          continue;
2,176,966✔
378
        }
379
      }
380

381
      // do apply filter
382
      code = doFilter(pFinalRes, pOperator->exprSupp.pFilterInfo, NULL);
1,147,679✔
383
      QUERY_CHECK_CODE(code, lino, _end);
1,147,791!
384

385
      // when apply the limit/offset for each group, pRes->info.rows may be 0, due to limit constraint.
386
      if (pFinalRes->info.rows > 0 || (pOperator->status == OP_EXEC_DONE)) {
1,147,791!
387
        qDebug("project return %" PRId64 " rows, status %d", pFinalRes->info.rows, pOperator->status);
1,147,940✔
388
        break;
1,147,760✔
389
      }
390
    } else {
391
      // do apply filter
392
      if (pRes->info.rows > 0) {
1,993,686✔
393
        code = doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);
1,311,782✔
394
        QUERY_CHECK_CODE(code, lino, _end);
1,311,838!
395

396
        if (pRes->info.rows == 0) {
1,311,838!
397
          continue;
×
398
        }
399
      }
400

401
      // no results generated
402
      break;
1,993,742✔
403
    }
404
  }
405

406
  SSDataBlock* p = pProjectInfo->mergeDataBlocks ? pFinalRes : pRes;
3,141,502✔
407
  pOperator->resultInfo.totalRows += p->info.rows;
3,141,502✔
408
  p->info.dataLoad = 1;
3,141,502✔
409

410
  if (pOperator->cost.openCost == 0) {
3,141,502✔
411
    pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
1,695,208✔
412
  }
413

414
  if (pProjectInfo->outputIgnoreGroup) {
3,141,553✔
415
    p->info.id.groupId = 0;
2,943,677✔
416
  }
417

418
  if (pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM) {
3,141,553✔
419
    printDataBlock(p, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
41,413✔
420
  }
421

422
  *pResBlock = (p->info.rows > 0)? p:NULL;
3,141,580✔
423

424
_end:
3,141,685✔
425
  if (code != TSDB_CODE_SUCCESS) {
3,141,685✔
426
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
105!
427
    pTaskInfo->code = code;
105✔
428
    T_LONG_JMP(pTaskInfo->env, code);
105!
429
  }
430
  return code;
3,141,580✔
431
}
432

433
int32_t createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode,
122,340✔
434
                                                 SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
435
  QRY_PARAM_CHECK(pOptrInfo);
122,340!
436
  int32_t code = 0;
122,340✔
437
  int32_t lino = 0;
122,340✔
438
  int32_t numOfRows = 4096;
122,340✔
439
  size_t  keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
122,340✔
440

441
  SIndefOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIndefOperatorInfo));
122,340✔
442
  SOperatorInfo*      pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
122,341✔
443
  if (pInfo == NULL || pOperator == NULL) {
122,338!
444
    code = terrno;
×
445
    goto _error;
×
446
  }
447

448
  pOperator->pTaskInfo = pTaskInfo;
122,338✔
449

450
  SExprSupp* pSup = &pOperator->exprSupp;
122,338✔
451
  pSup->hasWindowOrGroup = false;
122,338✔
452

453
  SIndefRowsFuncPhysiNode* pPhyNode = (SIndefRowsFuncPhysiNode*)pNode;
122,338✔
454

455
  if (pPhyNode->pExprs != NULL) {
122,338✔
456
    int32_t    num = 0;
139✔
457
    SExprInfo* pSExpr = NULL;
139✔
458
    code = createExprInfo(pPhyNode->pExprs, NULL, &pSExpr, &num);
139✔
459
    QUERY_CHECK_CODE(code, lino, _error);
139!
460

461
    code = initExprSupp(&pInfo->scalarSup, pSExpr, num, &pTaskInfo->storageAPI.functionStore);
139✔
462
    if (code != TSDB_CODE_SUCCESS) {
139!
463
      goto _error;
×
464
    }
465
  }
466

467
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->node.pOutputDataBlockDesc);
122,338✔
468
  TSDB_CHECK_NULL(pResBlock, code, lino, _error, terrno);
122,342!
469

470
  // Make sure the size of SSDataBlock will never exceed the size of 2MB.
471
  int32_t TWOMB = 2 * 1024 * 1024;
122,342✔
472
  if (numOfRows * pResBlock->info.rowSize > TWOMB) {
122,342!
473
    numOfRows = TWOMB / pResBlock->info.rowSize;
×
474
  }
475

476
  initBasicInfo(&pInfo->binfo, pResBlock);
122,342✔
477
  initResultSizeInfo(&pOperator->resultInfo, numOfRows);
122,342✔
478
  code = blockDataEnsureCapacity(pResBlock, numOfRows);
122,341✔
479
  TSDB_CHECK_CODE(code, lino, _error);
122,343!
480

481
  int32_t    numOfExpr = 0;
122,343✔
482
  SExprInfo* pExprInfo = NULL;
122,343✔
483
  code = createExprInfo(pPhyNode->pFuncs, NULL, &pExprInfo, &numOfExpr);
122,343✔
484
  TSDB_CHECK_CODE(code, lino, _error);
122,342!
485

486
  code = initAggSup(pSup, &pInfo->aggSup, pExprInfo, numOfExpr, keyBufSize, pTaskInfo->id.str,
122,342✔
487
                            pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);
122,342✔
488
  TSDB_CHECK_CODE(code, lino, _error);
122,344!
489

490
  code = setFunctionResultOutput(pOperator, &pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfExpr);
122,344✔
491
  TSDB_CHECK_CODE(code, lino, _error);
122,337!
492

493
  code = filterInitFromNode((SNode*)pPhyNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
122,337✔
494
  TSDB_CHECK_CODE(code, lino, _error);
122,339!
495

496
  pInfo->binfo.pRes = pResBlock;
122,339✔
497
  pInfo->binfo.inputTsOrder = pNode->inputTsOrder;
122,339✔
498
  pInfo->binfo.outputTsOrder = pNode->outputTsOrder;
122,339✔
499
  code = setRowTsColumnOutputInfo(pSup->pCtx, numOfExpr, &pInfo->pPseudoColInfo);
122,339✔
500
  TSDB_CHECK_CODE(code, lino, _error);
122,341!
501

502
  setOperatorInfo(pOperator, "IndefinitOperator", QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC, false, OP_NOT_OPENED, pInfo,
122,341✔
503
                  pTaskInfo);
504
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doApplyIndefinitFunction, NULL, destroyIndefinitOperatorInfo,
122,342✔
505
                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
506

507
  code = appendDownstream(pOperator, &downstream, 1);
122,339✔
508
  if (code != TSDB_CODE_SUCCESS) {
122,340!
509
    goto _error;
×
510
  }
511

512
  *pOptrInfo = pOperator;
122,340✔
513
  return TSDB_CODE_SUCCESS;
122,340✔
514

515
_error:
×
516
  if (pInfo != NULL) destroyIndefinitOperatorInfo(pInfo);
×
517
  destroyOperatorAndDownstreams(pOperator, &downstream, 1);
×
518
  pTaskInfo->code = code;
×
519
  return code;
×
520
}
521

522
static void doHandleDataBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, SOperatorInfo* downstream,
852,850✔
523
                              SExecTaskInfo* pTaskInfo) {
524
  SIndefOperatorInfo* pIndefInfo = pOperator->info;
852,850✔
525
  SOptrBasicInfo*     pInfo = &pIndefInfo->binfo;
852,850✔
526
  SExprSupp*          pSup = &pOperator->exprSupp;
852,850✔
527

528
  int32_t order = pInfo->inputTsOrder;
852,850✔
529
  int32_t scanFlag = pBlock->info.scanFlag;
852,850✔
530
  int32_t code = TSDB_CODE_SUCCESS;
852,850✔
531

532
  // there is an scalar expression that needs to be calculated before apply the group aggregation.
533
  SExprSupp* pScalarSup = &pIndefInfo->scalarSup;
852,850✔
534
  if (pScalarSup->pExprInfo != NULL) {
852,850✔
535
    code = projectApplyFunctions(pScalarSup->pExprInfo, pBlock, pBlock, pScalarSup->pCtx, pScalarSup->numOfExprs,
656✔
536
                                 pIndefInfo->pPseudoColInfo);
537
    if (code != TSDB_CODE_SUCCESS) {
656!
538
      T_LONG_JMP(pTaskInfo->env, code);
×
539
    }
540
  }
541

542
  code = setInputDataBlock(pSup, pBlock, order, scanFlag, false);
852,850✔
543
  if (code) {
852,847!
544
    T_LONG_JMP(pTaskInfo->env, code);
×
545
  }
546

547
  code = blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
852,847✔
548
  if (code != TSDB_CODE_SUCCESS) {
852,846!
549
    T_LONG_JMP(pTaskInfo->env, code);
×
550
  }
551

552
  code = projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs,
852,846✔
553
                               pIndefInfo->pPseudoColInfo);
554
  if (code != TSDB_CODE_SUCCESS) {
852,853✔
555
    T_LONG_JMP(pTaskInfo->env, code);
41!
556
  }
557
}
852,812✔
558

559
SSDataBlock* doApplyIndefinitFunction1(SOperatorInfo* pOperator) {
×
560
  SSDataBlock* pResBlock = NULL;
×
561
  pOperator->pTaskInfo->code = doApplyIndefinitFunction(pOperator, &pResBlock);
×
562
  return pResBlock;
×
563
}
564

565
int32_t doApplyIndefinitFunction(SOperatorInfo* pOperator, SSDataBlock** pResBlock) {
298,977✔
566
  QRY_PARAM_CHECK(pResBlock);
298,977!
567

568
  SIndefOperatorInfo* pIndefInfo = pOperator->info;
298,977✔
569
  SOptrBasicInfo*     pInfo = &pIndefInfo->binfo;
298,977✔
570
  SExprSupp*          pSup = &pOperator->exprSupp;
298,977✔
571
  int64_t             st = 0;
298,977✔
572
  int32_t             code = TSDB_CODE_SUCCESS;
298,977✔
573
  int32_t             lino = 0;
298,977✔
574
  SSDataBlock*        pRes = pInfo->pRes;
298,977✔
575

576
  blockDataCleanup(pRes);
298,977✔
577

578
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
298,977✔
579
  if (pOperator->status == OP_EXEC_DONE) {
298,977✔
580
    return code;
112,944✔
581
  }
582

583
  if (pOperator->cost.openCost == 0) {
186,033✔
584
    st = taosGetTimestampUs();
122,337✔
585
  }
586

587
  SOperatorInfo* downstream = pOperator->pDownstream[0];
186,037✔
588

589
  while (1) {
1,855✔
590
    // here we need to handle the existsed group results
591
    if (pIndefInfo->pNextGroupRes != NULL) {  // todo extract method
187,892✔
592
      for (int32_t k = 0; k < pSup->numOfExprs; ++k) {
212,051✔
593
        SqlFunctionCtx* pCtx = &pSup->pCtx[k];
157,706✔
594

595
        SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
157,706✔
596
        pResInfo->initialized = false;
157,706✔
597
        pCtx->pOutput = NULL;
157,706✔
598
      }
599

600
      doHandleDataBlock(pOperator, pIndefInfo->pNextGroupRes, downstream, pTaskInfo);
54,345✔
601
      pIndefInfo->pNextGroupRes = NULL;
54,345✔
602
    }
603

604
    if (pInfo->pRes->info.rows < pOperator->resultInfo.threshold) {
187,892✔
605
      while (1) {
787,260✔
606
        // The downstream exec may change the value of the newgroup, so use a local variable instead.
607
        SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);
975,149✔
608
        if (pBlock == NULL) {
975,147✔
609
          setOperatorCompleted(pOperator);
122,071✔
610
          break;
122,069✔
611
        }
612
        pInfo->pRes->info.scanFlag = pBlock->info.scanFlag;
853,076✔
613

614
        if (pIndefInfo->groupId == 0 && pBlock->info.id.groupId != 0) {
853,076✔
615
          pIndefInfo->groupId = pBlock->info.id.groupId;  // this is the initial group result
3,447✔
616
        } else {
617
          if (pIndefInfo->groupId != pBlock->info.id.groupId) {  // reset output buffer and computing status
849,629✔
618
            pIndefInfo->groupId = pBlock->info.id.groupId;
54,575✔
619
            pIndefInfo->pNextGroupRes = pBlock;
54,575✔
620
            break;
54,575✔
621
          }
622
        }
623

624
        doHandleDataBlock(pOperator, pBlock, downstream, pTaskInfo);
798,501✔
625
        if (pInfo->pRes->info.rows >= pOperator->resultInfo.threshold) {
798,467✔
626
          break;
11,207✔
627
        }
628
      }
629
    }
630

631
    code = doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
187,854✔
632
    QUERY_CHECK_CODE(code, lino, _end);
187,850!
633

634
    size_t rows = pInfo->pRes->info.rows;
187,850✔
635
    if (rows > 0 || pOperator->status == OP_EXEC_DONE) {
187,850✔
636
      break;
637
    } else {
638
      blockDataCleanup(pInfo->pRes);
1,855✔
639
    }
640
  }
641

642
  size_t rows = pInfo->pRes->info.rows;
185,995✔
643
  pOperator->resultInfo.totalRows += rows;
185,995✔
644

645
  if (pOperator->cost.openCost == 0) {
185,995✔
646
    pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
122,300✔
647
  }
648

649
  *pResBlock = (rows > 0) ? pInfo->pRes : NULL;
185,997✔
650

651
_end:
185,997✔
652
  if (code != TSDB_CODE_SUCCESS) {
185,997!
653
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
654
    pTaskInfo->code = code;
×
655
    T_LONG_JMP(pTaskInfo->env, code);
×
656
  }
657
  return code;
185,997✔
658
}
659

660
int32_t initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size) {
1,819,181✔
661
  int32_t code = TSDB_CODE_SUCCESS;
1,819,181✔
662
  for (int32_t j = 0; j < size; ++j) {
14,000,132✔
663
    struct SResultRowEntryInfo* pResInfo = GET_RES_INFO(&pCtx[j]);
12,180,474✔
664
    if (isRowEntryInitialized(pResInfo) || fmIsPseudoColumnFunc(pCtx[j].functionId) || pCtx[j].functionId == -1 ||
12,988,593!
665
        fmIsScalarFunc(pCtx[j].functionId)) {
808,157✔
666
      continue;
11,994,627✔
667
    }
668

669
    code = pCtx[j].fpSet.init(&pCtx[j], pCtx[j].resultInfo);
186,925✔
670
    if (code) {
186,324!
671
      return code;
×
672
    }
673
  }
674

675
  return 0;
1,819,658✔
676
}
677

678
/*
679
 * The start of each column SResultRowEntryInfo is denote by RowCellInfoOffset.
680
 * Note that in case of top/bottom query, the whole multiple rows of result is treated as only one row of results.
681
 * +------------+-----------------result column 1------------+------------------result column 2-----------+
682
 * | SResultRow | SResultRowEntryInfo | intermediate buffer1 | SResultRowEntryInfo | intermediate buffer 2|
683
 * +------------+--------------------------------------------+--------------------------------------------+
684
 *           offset[0]                                  offset[1]                                   offset[2]
685
 */
686
// TODO refactor: some function move away
687
int32_t setFunctionResultOutput(SOperatorInfo* pOperator, SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t stage,
1,818,819✔
688
                             int32_t numOfExprs) {
689
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
1,818,819✔
690
  SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;
1,818,819✔
691
  int32_t*        rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset;
1,818,819✔
692

693
  SResultRowInfo* pResultRowInfo = &pInfo->resultRowInfo;
1,818,819✔
694
  initResultRowInfo(pResultRowInfo);
1,818,819✔
695

696
  int64_t     tid = 0;
1,819,188✔
697
  int64_t     groupId = 0;
1,819,188✔
698
  SResultRow* pRow = doSetResultOutBufByKey(pSup->pResultBuf, pResultRowInfo, (char*)&tid, sizeof(tid), true, groupId,
1,819,188✔
699
                                            pTaskInfo, false, pSup, true);
700
  if (pRow == NULL || pTaskInfo->code != 0) {
1,819,587!
701
    return pTaskInfo->code;
×
702
  }
703

704
  for (int32_t i = 0; i < numOfExprs; ++i) {
13,998,539✔
705
    struct SResultRowEntryInfo* pEntry = getResultEntryInfo(pRow, i, rowEntryInfoOffset);
12,178,491✔
706
    cleanupResultRowEntry(pEntry);
12,178,952✔
707

708
    pCtx[i].resultInfo = pEntry;
12,178,952✔
709
    pCtx[i].scanFlag = stage;
12,178,952✔
710
  }
711

712
  return initCtxOutputBuffer(pCtx, numOfExprs);
1,820,048✔
713
}
714

715
int32_t setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols, SArray** pResList) {
1,818,244✔
716
  QRY_PARAM_CHECK(pResList);
1,818,244!
717
  SArray* pList = taosArrayInit(4, sizeof(int32_t));
1,818,244✔
718
  if (pList == NULL) {
1,819,870✔
719
    return terrno;
364✔
720
  }
721

722
  for (int32_t i = 0; i < numOfCols; ++i) {
14,003,145✔
723
    if (fmIsPseudoColumnFunc(pCtx[i].functionId)) {
12,183,168!
724
      void* px = taosArrayPush(pList, &i);
×
725
      if (px == NULL) {
×
726
        return terrno;
×
727
      }
728
    }
729
  }
730

731
  *pResList = pList;
1,819,977✔
732
  return 0;
1,819,977✔
733
}
734

735
int32_t doGenerateSourceData(SOperatorInfo* pOperator) {
1,591✔
736
  SProjectOperatorInfo* pProjectInfo = pOperator->info;
1,591✔
737

738
  SExprSupp*   pSup = &pOperator->exprSupp;
1,591✔
739
  SSDataBlock* pRes = pProjectInfo->binfo.pRes;
1,591✔
740
  SExprInfo*   pExpr = pSup->pExprInfo;
1,591✔
741
  int64_t      st = taosGetTimestampUs();
1,591✔
742
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1,591✔
743

744
  int32_t code = blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
1,591✔
745
  if (code) {
1,591!
746
    return code;
×
747
  }
748

749
  for (int32_t k = 0; k < pSup->numOfExprs; ++k) {
3,186✔
750
    int32_t outputSlotId = pExpr[k].base.resSchema.slotId;
1,595✔
751

752
    if (pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE) {
1,595✔
753
      SColumnInfoData* pColInfoData = taosArrayGet(pRes->pDataBlock, outputSlotId);
1,591✔
754
      if (pColInfoData == NULL) {
1,591!
755
        return terrno;
×
756
      }
757

758
      int32_t type = pExpr[k].base.pParam[0].param.nType;
1,591✔
759
      if (TSDB_DATA_TYPE_NULL == type) {
1,591✔
760
        colDataSetNNULL(pColInfoData, 0, 1);
761
      } else {
762
        code = colDataSetVal(pColInfoData, 0, taosVariantGet(&pExpr[k].base.pParam[0].param, type), false);
1,589✔
763
        if (code) {
1,589!
764
          return code;
×
765
        }
766
      }
767
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
4!
768
      SqlFunctionCtx* pfCtx = &pSup->pCtx[k];
4✔
769

770
      // UDF scalar functions will be calculated here, for example, select foo(n) from (select 1 n).
771
      // UDF aggregate functions will be handled in agg operator.
772
      if (fmIsScalarFunc(pfCtx->functionId)) {
4!
773
        SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
4✔
774
        if (pBlockList == NULL) {
4!
775
          return terrno;
×
776
        }
777

778
        void* px = taosArrayPush(pBlockList, &pRes);
4✔
779
        if (px == NULL) {
4!
780
          return terrno;
×
781
        }
782

783
        SColumnInfoData* pResColData = taosArrayGet(pRes->pDataBlock, outputSlotId);
4✔
784
        if (pResColData == NULL) {
4!
785
          return terrno;
×
786
        }
787

788
        SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
4✔
789

790
        SScalarParam dest = {.columnData = &idata};
4✔
791
        code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
4✔
792
        if (code != TSDB_CODE_SUCCESS) {
4!
793
          taosArrayDestroy(pBlockList);
×
794
          return code;
×
795
        }
796

797
        int32_t startOffset = pRes->info.rows;
4✔
798
        if (pRes->info.capacity <= 0) {
4!
799
          qError("project failed at: %s:%d", __func__, __LINE__);
×
800
          return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
801
        }
802
        code = colDataAssign(pResColData, &idata, dest.numOfRows, &pRes->info);
4✔
803
        if (code) {
4!
804
          return code;
×
805
        }
806

807
        colDataDestroy(&idata);
4✔
808
        taosArrayDestroy(pBlockList);
4✔
809
      } else {
810
        return TSDB_CODE_OPS_NOT_SUPPORT;
×
811
      }
812
    } else {
813
      return TSDB_CODE_OPS_NOT_SUPPORT;
×
814
    }
815
  }
816

817
  pRes->info.rows = 1;
1,591✔
818
  code = doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);
1,591✔
819
  if (code) {
1,591!
820
    pTaskInfo->code = code;
×
821
    return code;
×
822
  }
823

824
  (void) doIngroupLimitOffset(&pProjectInfo->limitInfo, 0, pRes, pOperator);
1,591✔
825

826
  pOperator->resultInfo.totalRows += pRes->info.rows;
1,591✔
827

828
  setOperatorCompleted(pOperator);
1,591✔
829
  if (pOperator->cost.openCost == 0) {
1,591!
830
    pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
1,591✔
831
  }
832

833
  return code;
1,591✔
834
}
835

836
static void setPseudoOutputColInfo(SSDataBlock* pResult, SqlFunctionCtx* pCtx, SArray* pPseudoList) {
6,242,230✔
837
  size_t num = (pPseudoList != NULL) ? taosArrayGetSize(pPseudoList) : 0;
6,242,230✔
838
  for (int32_t i = 0; i < num; ++i) {
6,243,261!
839
    pCtx[i].pOutput = taosArrayGet(pResult->pDataBlock, i);
×
840
    if (pCtx[i].pOutput == NULL) {
×
841
      qError("failed to get the output buf, ptr is null");
×
842
    }
843
  }
844
}
6,243,261✔
845

846
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
6,242,527✔
847
                              int32_t numOfOutput, SArray* pPseudoList) {
848
  int32_t lino = 0;
6,242,527✔
849
  int32_t code = TSDB_CODE_SUCCESS;
6,242,527✔
850
  setPseudoOutputColInfo(pResult, pCtx, pPseudoList);
6,242,527✔
851
  pResult->info.dataLoad = 1;
6,243,335✔
852

853
  SArray* processByRowFunctionCtx = NULL;
6,243,335✔
854
  if (pSrcBlock == NULL) {
6,243,335!
855
    for (int32_t k = 0; k < numOfOutput; ++k) {
×
856
      int32_t outputSlotId = pExpr[k].base.resSchema.slotId;
×
857

858
      if (pExpr[k].pExpr->nodeType != QUERY_NODE_VALUE) {
×
859
        qError("project failed at: %s:%d", __func__, __LINE__);
×
860
        code = TSDB_CODE_INVALID_PARA;
×
861
        TSDB_CHECK_CODE(code, lino, _exit);
×
862
      }
863
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
×
864
      if (pColInfoData == NULL) {
×
865
        code = terrno;
×
866
        TSDB_CHECK_CODE(code, lino, _exit);
×
867
      }
868

869
      int32_t type = pExpr[k].base.pParam[0].param.nType;
×
870
      if (TSDB_DATA_TYPE_NULL == type) {
×
871
        colDataSetNNULL(pColInfoData, 0, 1);
872
      } else {
873
        code = colDataSetVal(pColInfoData, 0, taosVariantGet(&pExpr[k].base.pParam[0].param, type), false);
×
874
        TSDB_CHECK_CODE(code, lino, _exit);
×
875
      }
876
    }
877

878
    pResult->info.rows = 1;
×
879
    goto _exit;
×
880
  }
881

882
  if (pResult != pSrcBlock) {
6,243,335✔
883
    pResult->info.id.groupId = pSrcBlock->info.id.groupId;
5,106,423✔
884
    memcpy(pResult->info.parTbName, pSrcBlock->info.parTbName, TSDB_TABLE_NAME_LEN);
5,106,423✔
885
  }
886

887
  // if the source equals to the destination, it is to create a new column as the result of scalar
888
  // function or some operators.
889
  bool createNewColModel = (pResult == pSrcBlock);
6,243,335✔
890
  if (createNewColModel) {
6,243,335✔
891
    code = blockDataEnsureCapacity(pResult, pResult->info.rows);
1,135,595✔
892
    if (code) {
1,135,920!
893
      goto _exit;
×
894
    }
895
  }
896

897
  int32_t numOfRows = 0;
6,243,660✔
898

899
  for (int32_t k = 0; k < numOfOutput; ++k) {
30,726,993✔
900
    int32_t               outputSlotId = pExpr[k].base.resSchema.slotId;
24,618,467✔
901
    SqlFunctionCtx*       pfCtx = &pCtx[k];
24,618,467✔
902
    SInputColumnInfoData* pInputData = &pfCtx->input;
24,618,467✔
903

904
    if (pExpr[k].pExpr->nodeType == QUERY_NODE_COLUMN) {  // it is a project query
24,618,467✔
905
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
18,254,648✔
906
      if (pColInfoData == NULL) {
18,253,116✔
907
        code = terrno;
228✔
908
        TSDB_CHECK_CODE(code, lino, _exit);
174!
909
      }
910

911
      if (pResult->info.rows > 0 && !createNewColModel) {
18,252,888!
912
        int32_t ret = 0;
×
913

914
        if (pInputData->pData[0] == NULL) {
×
915
          int32_t slotId = pfCtx->param[0].pCol->slotId;
×
916

917
          SColumnInfoData* pInput = taosArrayGet(pSrcBlock->pDataBlock, slotId);
×
918
          if (pInput == NULL) {
×
919
            code = terrno;
×
920
            TSDB_CHECK_CODE(code, lino, _exit);
×
921
          }
922

923
          ret = colDataMergeCol(pColInfoData, pResult->info.rows, (int32_t*)&pResult->info.capacity, pInput,
×
924
                                pSrcBlock->info.rows);
×
925
        } else {
926
          ret = colDataMergeCol(pColInfoData, pResult->info.rows, (int32_t*)&pResult->info.capacity,
×
927
                                pInputData->pData[0], pInputData->numOfRows);
×
928
        }
929

930
        if (ret < 0) {
×
931
          code = ret;
×
932
        }
933

934
        TSDB_CHECK_CODE(code, lino, _exit);
×
935
      } else {
936
        if (pInputData->pData[0] == NULL) {
18,252,888✔
937
          int32_t slotId = pfCtx->param[0].pCol->slotId;
24,801✔
938

939
          SColumnInfoData* pInput = taosArrayGet(pSrcBlock->pDataBlock, slotId);
24,801✔
940
          if (pInput == NULL) {
24,801!
941
            code = terrno;
×
942
            TSDB_CHECK_CODE(code, lino, _exit);
×
943
          }
944

945
          code = colDataAssign(pColInfoData, pInput, pSrcBlock->info.rows, &pResult->info);
24,801✔
946
          numOfRows = pSrcBlock->info.rows;
24,801✔
947
        } else {
948
          code = colDataAssign(pColInfoData, pInputData->pData[0], pInputData->numOfRows, &pResult->info);
18,228,087✔
949
          numOfRows = pInputData->numOfRows;
18,228,863✔
950
        }
951

952
        TSDB_CHECK_CODE(code, lino, _exit);
18,253,664!
953
      }
954
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE) {
6,363,819✔
955
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
482,823✔
956
      if (pColInfoData == NULL) {
482,880!
957
        code = terrno;
×
958
        TSDB_CHECK_CODE(code, lino, _exit);
×
959
      }
960

961
      int32_t offset = createNewColModel ? 0 : pResult->info.rows;
482,885!
962

963
      int32_t type = pExpr[k].base.pParam[0].param.nType;
482,885✔
964
      if (TSDB_DATA_TYPE_NULL == type) {
482,885✔
965
        colDataSetNNULL(pColInfoData, offset, pSrcBlock->info.rows);
23,268✔
966
      } else {
967
        char* p = taosVariantGet(&pExpr[k].base.pParam[0].param, type);
459,617✔
968
        for (int32_t i = 0; i < pSrcBlock->info.rows; ++i) {
346,254,921✔
969
          code = colDataSetVal(pColInfoData, i + offset, p, false);
345,932,022✔
970
          TSDB_CHECK_CODE(code, lino, _exit);
345,795,327!
971
        }
972
      }
973

974
      numOfRows = pSrcBlock->info.rows;
346,167✔
975
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_OPERATOR) {
5,880,996✔
976
      SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
416,025✔
977
      if (pBlockList == NULL) {
416,300!
978
        code = terrno;
×
979
        goto _exit;
73✔
980
      }
981

982
      void* px = taosArrayPush(pBlockList, &pSrcBlock);
416,171✔
983
      if (px == NULL) {
416,171!
984
        code = terrno;
×
985
        taosArrayDestroy(pBlockList);
×
986
        goto _exit;
×
987
      }
988

989
      SColumnInfoData* pResColData = taosArrayGet(pResult->pDataBlock, outputSlotId);
416,171✔
990
      if (pResColData == NULL) {
416,143!
991
        code = terrno;
×
992
        taosArrayDestroy(pBlockList);
×
993
        goto _exit;
×
994
      }
995

996
      SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
416,143✔
997

998
      SScalarParam dest = {.columnData = &idata};
416,143✔
999
      code = scalarCalculate(pExpr[k].pExpr->_optrRoot.pRootNode, pBlockList, &dest);
416,143✔
1000
      if (code != TSDB_CODE_SUCCESS) {
415,978✔
1001
        taosArrayDestroy(pBlockList);
105✔
1002
        goto _exit;
73✔
1003
      }
1004

1005
      int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
415,873✔
1006
      if (pResult->info.capacity <= 0) {
415,873!
1007
        qError("project failed at: %s:%d", __func__, __LINE__);
×
1008
        code = TSDB_CODE_INVALID_PARA;
×
1009
        TSDB_CHECK_CODE(code, lino, _exit);
×
1010
      }
1011

1012
      int32_t ret = colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
415,873✔
1013
      if (ret < 0) {
415,676!
1014
        code = ret;
×
1015
      }
1016

1017
      colDataDestroy(&idata);
415,676✔
1018
      TSDB_CHECK_CODE(code, lino, _exit);
416,021!
1019

1020
      numOfRows = dest.numOfRows;
416,021✔
1021
      taosArrayDestroy(pBlockList);
416,021✔
1022
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
5,464,971✔
1023
      // _rowts/_c0, not tbname column
1024
      if (fmIsPseudoColumnFunc(pfCtx->functionId) && (!fmIsScanPseudoColumnFunc(pfCtx->functionId))) {
5,464,299!
1025
        // do nothing
1026
      } else if (fmIsIndefiniteRowsFunc(pfCtx->functionId)) {
5,464,413✔
1027
        SResultRowEntryInfo* pResInfo = GET_RES_INFO(pfCtx);
853,227✔
1028
        code = pfCtx->fpSet.init(pfCtx, pResInfo);
853,227✔
1029
        TSDB_CHECK_CODE(code, lino, _exit);
853,230!
1030
        pfCtx->pOutput = taosArrayGet(pResult->pDataBlock, outputSlotId);
853,230✔
1031
        if (pfCtx->pOutput == NULL) {
853,229!
1032
          code = terrno;
×
1033
          TSDB_CHECK_CODE(code, lino, _exit);
×
1034
        }
1035

1036
        pfCtx->offset = createNewColModel ? 0 : pResult->info.rows;  // set the start offset
853,229!
1037

1038
        // set the timestamp(_rowts) output buffer
1039
        if (taosArrayGetSize(pPseudoList) > 0) {
853,229!
1040
          int32_t* outputColIndex = taosArrayGet(pPseudoList, 0);
×
1041
          if (outputColIndex == NULL) {
×
1042
            code = terrno;
×
1043
            goto _exit;
×
1044
          }
1045

1046
          pfCtx->pTsOutput = (SColumnInfoData*)pCtx[*outputColIndex].pOutput;
×
1047
        }
1048

1049
        // link pDstBlock to set selectivity value
1050
        if (pfCtx->subsidiaries.num > 0) {
853,230✔
1051
          pfCtx->pDstBlock = pResult;
628,646✔
1052
        }
1053

1054
        code = pfCtx->fpSet.process(pfCtx);
853,230✔
1055
        if (code != TSDB_CODE_SUCCESS) {
853,232✔
1056
          if (pCtx[k].fpSet.cleanup != NULL) {
18!
1057
            pCtx[k].fpSet.cleanup(&pCtx[k]);
×
1058
          }
1059
          goto _exit;
18✔
1060
        }
1061

1062
        numOfRows = pResInfo->numOfRes;
853,214✔
1063
        if (fmIsProcessByRowFunc(pfCtx->functionId)) {
853,214✔
1064
          if (NULL == processByRowFunctionCtx) {
411,127✔
1065
            processByRowFunctionCtx = taosArrayInit(1, sizeof(SqlFunctionCtx*));
410,904✔
1066
            if (!processByRowFunctionCtx) {
410,904!
1067
              code = terrno;
×
1068
              goto _exit;
×
1069
            }
1070
          }
1071

1072
          void* px = taosArrayPush(processByRowFunctionCtx, &pfCtx);
411,162✔
1073
          if (px == NULL) {
411,162!
1074
            code = terrno;
×
1075
            goto _exit;
×
1076
          }
1077
        }
1078
      } else if (fmIsAggFunc(pfCtx->functionId)) {
4,608,998✔
1079
        // selective value output should be set during corresponding function execution
1080
        if (fmIsSelectValueFunc(pfCtx->functionId)) {
990,514✔
1081
          continue;
622,390✔
1082
        }
1083
        // _group_key function for "partition by tbname" + csum(col_name) query
1084
        SColumnInfoData* pOutput = taosArrayGet(pResult->pDataBlock, outputSlotId);
368,124✔
1085
        if (pOutput == NULL) {
368,124!
1086
          code = terrno;
×
1087
          TSDB_CHECK_CODE(code, lino, _exit);
×
1088
        }
1089

1090
        int32_t          slotId = pfCtx->param[0].pCol->slotId;
368,124✔
1091

1092
        // todo handle the json tag
1093
        SColumnInfoData* pInput = taosArrayGet(pSrcBlock->pDataBlock, slotId);
368,124✔
1094
        if (pInput == NULL) {
368,124!
1095
          code = terrno;
×
1096
          TSDB_CHECK_CODE(code, lino, _exit);
×
1097
        }
1098

1099
        for (int32_t f = 0; f < pSrcBlock->info.rows; ++f) {
26,590,557✔
1100
          bool isNull = colDataIsNull_s(pInput, f);
26,222,433✔
1101
          if (isNull) {
26,222,433✔
1102
            colDataSetNULL(pOutput, pResult->info.rows + f);
2,586,150✔
1103
          } else {
1104
            char* data = colDataGetData(pInput, f);
23,636,283!
1105
            code = colDataSetVal(pOutput, pResult->info.rows + f, data, isNull);
23,636,283✔
1106
            TSDB_CHECK_CODE(code, lino, _exit);
23,636,283!
1107
          }
1108
        }
1109

1110
      } else {
1111
        SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
3,615,962✔
1112
        if (pBlockList == NULL) {
3,623,764!
1113
          code = terrno;
×
1114
          goto _exit;
83✔
1115
        }
1116

1117
        void* px = taosArrayPush(pBlockList, &pSrcBlock);
3,622,339✔
1118
        if (px == NULL) {
3,622,339!
1119
          code = terrno;
×
1120
          goto _exit;
×
1121
        }
1122

1123
        SColumnInfoData* pResColData = taosArrayGet(pResult->pDataBlock, outputSlotId);
3,622,339✔
1124
        if (pResColData == NULL) {
3,622,339!
1125
          taosArrayDestroy(pBlockList);
×
1126
          code = terrno;
×
1127
          goto _exit;
×
1128
        }
1129

1130
        SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
3,622,339✔
1131

1132
        SScalarParam dest = {.columnData = &idata};
3,622,339✔
1133
        code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
3,622,339✔
1134
        if (code != TSDB_CODE_SUCCESS) {
3,620,848✔
1135
          taosArrayDestroy(pBlockList);
203✔
1136
          goto _exit;
83✔
1137
        }
1138

1139
        int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
3,620,645✔
1140
        if (pResult->info.capacity <= 0) {
3,620,645!
1141
          qError("project failed at: %s:%d", __func__, __LINE__);
×
1142
          code = TSDB_CODE_INVALID_PARA;
×
1143
          TSDB_CHECK_CODE(code, lino, _exit);
×
1144
        }
1145
        int32_t ret = colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
3,620,645✔
1146
        if (ret < 0) {
3,617,990!
1147
          code = ret;
×
1148
        }
1149

1150
        colDataDestroy(&idata);
3,617,990✔
1151

1152
        numOfRows = dest.numOfRows;
3,623,784✔
1153
        taosArrayDestroy(pBlockList);
3,623,784✔
1154
        TSDB_CHECK_CODE(code, lino, _exit);
3,623,652!
1155
      }
1156
    } else {
1157
      return TSDB_CODE_OPS_NOT_SUPPORT;
672✔
1158
    }
1159
  }
1160

1161
  if (processByRowFunctionCtx && taosArrayGetSize(processByRowFunctionCtx) > 0){
6,108,526✔
1162
    SqlFunctionCtx** pfCtx = taosArrayGet(processByRowFunctionCtx, 0);
275,809✔
1163
    if (pfCtx == NULL) {
410,904!
1164
      code = terrno;
×
1165
      goto _exit;
×
1166
    }
1167

1168
    code = (*pfCtx)->fpSet.processFuncByRow(processByRowFunctionCtx);
410,904✔
1169
    TSDB_CHECK_CODE(code, lino, _exit);
410,904✔
1170
    numOfRows = (*pfCtx)->resultInfo->numOfRes;
410,881✔
1171
  }
1172

1173
  if (!createNewColModel) {
6,243,598✔
1174
    pResult->info.rows += numOfRows;
5,106,929✔
1175
  }
1176

1177
_exit:
1,136,669✔
1178
  if(processByRowFunctionCtx) {
6,243,795✔
1179
    taosArrayDestroy(processByRowFunctionCtx);
410,904✔
1180
  }
1181
  return code;
6,244,035✔
1182
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc