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

taosdata / TDengine / #4917

07 Jan 2026 03:52PM UTC coverage: 65.42% (+0.02%) from 65.402%
#4917

push

travis-ci

web-flow
merge: from main to 3.0 branch #34204

31 of 34 new or added lines in 2 files covered. (91.18%)

819 existing lines in 129 files now uncovered.

202679 of 309814 relevant lines covered (65.42%)

116724351.99 hits per line

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

60.18
/source/libs/executor/src/operator.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 "filter.h"
17
#include "function.h"
18
#include "tname.h"
19

20
#include "tglobal.h"
21

22
#include "executorInt.h"
23
#include "index.h"
24
#include "operator.h"
25
#include "query.h"
26
#include "querytask.h"
27

28
#include "storageapi.h"
29
#include "tdatablock.h"
30

31
SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
748,092,580✔
32
                                   __optr_close_fn_t closeFn, __optr_reqBuf_fn_t reqBufFn, __optr_explain_fn_t explain,
33
                                   __optr_get_ext_fn_t nextExtFn, __optr_notify_fn_t notifyFn) {
34
  SOperatorFpSet fpSet = {
748,092,580✔
35
      ._openFn = openFn,
36
      .getNextFn = nextFn,
37
      .cleanupFn = cleanup,
38
      .closeFn = closeFn,
39
      .reqBufFn = reqBufFn,
40
      .getExplainFn = explain,
41
      .getNextExtFn = nextExtFn,
42
      .notifyFn = notifyFn,
43
      .releaseStreamStateFn = NULL,
44
      .reloadStreamStateFn = NULL,
45
      .resetStateFn = NULL,
46
  };
47

48
  return fpSet;
748,092,580✔
49
}
50

51
void setOperatorStreamStateFn(SOperatorInfo* pOperator, __optr_state_fn_t relaseFn, __optr_state_fn_t reloadFn) {
129,979,810✔
52
  pOperator->fpSet.releaseStreamStateFn = relaseFn;
129,979,810✔
53
  pOperator->fpSet.reloadStreamStateFn = reloadFn;
129,997,564✔
54
}
129,915,658✔
55

56
void setOperatorResetStateFn(SOperatorInfo* pOperator, __optr_reset_state_fn_t resetFn) {
746,768,149✔
57
  pOperator->fpSet.resetStateFn = resetFn;
746,768,149✔
58
}
746,957,472✔
59

60
int32_t optrDummyOpenFn(SOperatorInfo* pOperator) {
63,942,116✔
61
  OPTR_SET_OPENED(pOperator);
63,942,116✔
62
  pOperator->cost.openCost = 0;
63,944,316✔
63
  return TSDB_CODE_SUCCESS;
63,948,761✔
64
}
65

66
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num) {
361,136,187✔
67
  p->pDownstream = taosMemoryCalloc(1, num * POINTER_BYTES);
361,136,187✔
68
  if (p->pDownstream == NULL) {
361,095,302✔
69
    return terrno;
×
70
  }
71

72
  memcpy(p->pDownstream, pDownstream, num * POINTER_BYTES);
361,017,446✔
73
  p->numOfDownstream = num;
361,056,941✔
74
  p->numOfRealDownstream = num;
361,247,588✔
75
  return TSDB_CODE_SUCCESS;
361,051,881✔
76
}
77

78
void setOperatorCompleted(SOperatorInfo* pOperator) {
716,813,102✔
79
  pOperator->status = OP_EXEC_DONE;
716,813,102✔
80
  pOperator->cost.totalCost = (taosGetTimestampUs() - pOperator->pTaskInfo->cost.start) / 1000.0;
716,963,287✔
81
  setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
716,947,275✔
82
}
716,914,454✔
83

84
void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
747,907,142✔
85
                     void* pInfo, SExecTaskInfo* pTaskInfo) {
86
  pOperator->name = (char*)name;
747,907,142✔
87
  pOperator->operatorType = type;
748,174,439✔
88
  pOperator->blocking = blocking;
747,543,713✔
89
  pOperator->status = status;
747,969,306✔
90
  pOperator->info = pInfo;
747,866,758✔
91
  pOperator->pTaskInfo = pTaskInfo;
747,965,674✔
92
}
747,802,032✔
93

94
// each operator should be set their own function to return total cost buffer
95
int32_t optrDefaultBufFn(SOperatorInfo* pOperator) {
×
96
  if (pOperator->blocking) {
×
97
    return -1;
×
98
  } else {
99
    return 0;
×
100
  }
101
}
102

103
static int64_t getQuerySupportBufSize(size_t numOfTables) {
×
104
  size_t s1 = sizeof(STableQueryInfo);
×
105
  //  size_t s3 = sizeof(STableCheckInfo);  buffer consumption in tsdb
106
  return (int64_t)(s1 * 1.5 * numOfTables);
×
107
}
108

109
int32_t checkForQueryBuf(size_t numOfTables) {
×
110
  int64_t t = getQuerySupportBufSize(numOfTables);
×
111
  if (tsQueryBufferSizeBytes < 0) {
×
112
    return TSDB_CODE_SUCCESS;
×
113
  } else if (tsQueryBufferSizeBytes > 0) {
×
114
    while (1) {
×
115
      int64_t s = tsQueryBufferSizeBytes;
×
116
      int64_t remain = s - t;
×
117
      if (remain >= 0) {
×
118
        if (atomic_val_compare_exchange_64(&tsQueryBufferSizeBytes, s, remain) == s) {
×
119
          return TSDB_CODE_SUCCESS;
×
120
        }
121
      } else {
122
        return TSDB_CODE_QRY_NOT_ENOUGH_BUFFER;
×
123
      }
124
    }
125
  }
126

127
  // disable query processing if the value of tsQueryBufferSize is zero.
128
  return TSDB_CODE_QRY_NOT_ENOUGH_BUFFER;
×
129
}
130

131
void releaseQueryBuf(size_t numOfTables) {
×
132
  if (tsQueryBufferSizeBytes < 0) {
×
133
    return;
×
134
  }
135

136
  int64_t t = getQuerySupportBufSize(numOfTables);
×
137

138
  // restore value is not enough buffer available
139
  (void)atomic_add_fetch_64(&tsQueryBufferSizeBytes, t);
×
140
}
141

142
typedef enum {
143
  OPTR_FN_RET_CONTINUE = 0x1,
144
  OPTR_FN_RET_ABORT = 0x2,
145
} ERetType;
146

147
typedef struct STraverParam {
148
  void*   pRet;
149
  int32_t code;
150
  void*   pParam;
151
} STraverParam;
152

153
// iterate the operator tree helper
154
typedef ERetType (*optr_fn_t)(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdstr);
155

156
void traverseOperatorTree(SOperatorInfo* pOperator, optr_fn_t fn, STraverParam* pParam, const char* id) {
45,360,584✔
157
  if (pOperator == NULL) {
45,360,584✔
158
    return;
×
159
  }
160

161
  ERetType ret = fn(pOperator, pParam, id);
45,360,584✔
162
  if (ret == OPTR_FN_RET_ABORT || pParam->code != TSDB_CODE_SUCCESS) {
45,360,975✔
163
    return;
23,403,309✔
164
  }
165

166
  for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
43,866,489✔
167
    traverseOperatorTree(pOperator->pDownstream[i], fn, pParam, id);
21,910,404✔
168
    if (pParam->code != 0) {
21,908,281✔
169
      break;
×
170
    }
171
  }
172
}
173

174
ERetType extractOperatorInfo(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
44,932,168✔
175
  STraverParam* p = pParam;
44,932,168✔
176
  if (pOperator->operatorType == *(int32_t*)p->pParam) {
44,932,168✔
177
    p->pRet = pOperator;
23,246,451✔
178
    return OPTR_FN_RET_ABORT;
23,246,451✔
179
  } else {
180
    return OPTR_FN_RET_CONTINUE;
21,686,373✔
181
  }
182
}
183

184
// QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN
185
int32_t extractOperatorInTree(SOperatorInfo* pOperator, int32_t type, const char* id, SOperatorInfo** pOptrInfo) {
23,246,724✔
186
  QRY_PARAM_CHECK(pOptrInfo);
23,246,724✔
187

188
  if (pOperator == NULL) {
23,247,680✔
189
    qError("invalid operator, failed to find tableScanOperator %s", id);
×
190
    return TSDB_CODE_PAR_INTERNAL_ERROR;
×
191
  }
192

193
  STraverParam p = {.pParam = &type, .pRet = NULL};
23,247,680✔
194
  traverseOperatorTree(pOperator, extractOperatorInfo, &p, id);
23,247,051✔
195
  if (p.code == 0) {
23,245,166✔
196
    *pOptrInfo = p.pRet;
23,246,122✔
197
  }
198
  return p.code;
23,245,491✔
199
}
200

201
typedef struct SExtScanInfo {
202
  int32_t order;
203
  int32_t scanFlag;
204
  int32_t inheritUsOrder;
205
} SExtScanInfo;
206

207
static ERetType extractScanInfo(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
×
208
  int32_t       type = pOperator->operatorType;
×
209
  SExtScanInfo* pInfo = pParam->pParam;
×
210

211
  if (type == QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN ||
×
212
      type == QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN ||
×
213
      type == QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN) {
×
214
    pInfo->order = TSDB_ORDER_ASC;
×
215
    pInfo->scanFlag = MAIN_SCAN;
×
216
    return OPTR_FN_RET_ABORT;
×
217
  } else if (type == QUERY_NODE_PHYSICAL_PLAN_EXCHANGE) {
×
218
    if (!pInfo->inheritUsOrder) {
×
219
      pInfo->order = TSDB_ORDER_ASC;
×
220
    }
221
    pInfo->scanFlag = MAIN_SCAN;
×
222
    return OPTR_FN_RET_ABORT;
×
223
  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
×
224
    STableScanInfo* pTableScanInfo = pOperator->info;
×
225
    pInfo->order = pTableScanInfo->base.cond.order;
×
226
    pInfo->scanFlag = pTableScanInfo->base.scanFlag;
×
227
    return OPTR_FN_RET_ABORT;
×
228
  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN) {
×
229
    STableMergeScanInfo* pTableScanInfo = pOperator->info;
×
230
    pInfo->order = pTableScanInfo->base.cond.order;
×
231
    pInfo->scanFlag = pTableScanInfo->base.scanFlag;
×
232
    return OPTR_FN_RET_ABORT;
×
233
  } else {
234
    return OPTR_FN_RET_CONTINUE;
×
235
  }
236
}
237

238
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag, bool inheritUsOrder) {
×
239
  SExtScanInfo info = {.inheritUsOrder = inheritUsOrder, .order = *order};
×
240
  STraverParam p = {.pParam = &info};
×
241

242
  traverseOperatorTree(pOperator, extractScanInfo, &p, NULL);
×
243
  *order = info.order;
×
244
  *scanFlag = info.scanFlag;
×
245

246
  if (p.code == TSDB_CODE_SUCCESS) {
×
247
    if (!(*order == TSDB_ORDER_ASC || *order == TSDB_ORDER_DESC)) {
×
248
      qError("operator failed at: %s:%d", __func__, __LINE__);
×
249
      p.code = TSDB_CODE_INVALID_PARA;
×
250
    }
251
  }
252
  return p.code;
×
253
}
254

255
static ERetType doStopDataReader(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
428,393✔
256
  SStorageAPI* pAPI = pParam->pParam;
428,393✔
257
  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
428,393✔
258
    STableScanInfo* pInfo = pOperator->info;
157,752✔
259

260
    if (pInfo->base.dataReader != NULL) {
157,752✔
261
      pAPI->tsdReader.tsdReaderNotifyClosing(pInfo->base.dataReader);
156,393✔
262
    }
263
    return OPTR_FN_RET_ABORT;
157,752✔
264
  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
270,641✔
265
    SStreamScanInfo* pInfo = pOperator->info;
×
266

267
    if (pInfo->pTableScanOp != NULL) {
×
268
      STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
×
269
      if (pTableScanInfo != NULL && pTableScanInfo->base.dataReader != NULL) {
×
270
        pAPI->tsdReader.tsdReaderNotifyClosing(pTableScanInfo->base.dataReader);
×
271
      }
272
    }
273

274
    return OPTR_FN_RET_ABORT;
×
275
  }
276

277
  return OPTR_FN_RET_CONTINUE;
270,641✔
278
}
279

280
int32_t stopTableScanOperator(SOperatorInfo* pOperator, const char* pIdStr, SStorageAPI* pAPI) {
204,991✔
281
  STraverParam p = {.pParam = pAPI};
204,991✔
282
  traverseOperatorTree(pOperator, doStopDataReader, &p, pIdStr);
204,991✔
283
  return p.code;
204,991✔
284
}
285

286
int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SNode* pTagCond,
756,803,119✔
287
                       SNode* pTagIndexCond, const char* pUser, const char* dbname, SOperatorInfo** pOptrInfo,
288
                       EOPTR_EXEC_MODEL model) {
289
  QRY_PARAM_CHECK(pOptrInfo);
756,803,119✔
290

291
  int32_t     code = 0;
756,874,965✔
292
  int32_t     type = nodeType(pPhyNode);
756,874,965✔
293
  const char* idstr = GET_TASKID(pTaskInfo);
756,874,540✔
294

295
  if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
756,830,410✔
296
    SOperatorInfo* pOperator = NULL;
386,571,653✔
297
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
386,634,827✔
298
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
235,466,341✔
299
      // NOTE: this is an patch to fix the physical plan
300
      // TODO remove it later
301
      if (pTableScanNode->scan.node.pLimit != NULL) {
235,466,341✔
302
        pTableScanNode->groupSort = true;
4,338,076✔
303
      }
304

305
      STableListInfo* pTableListInfo = tableListCreate();
235,490,870✔
306
      if (!pTableListInfo) {
235,581,198✔
307
        pTaskInfo->code = terrno;
×
308
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
309
        return terrno;
×
310
      }
311

312
      // Since virtual stable scan use virtual super table's uid in scan operator, the origin table might be stored on
313
      // different vnode, so we should not get table schema for virtual stable scan.
314
      if (!pTableScanNode->scan.virtualStableScan) {
235,581,198✔
315
        code = initQueriedTableSchemaInfo(pHandle, &pTableScanNode->scan, dbname, pTaskInfo);
235,379,039✔
316
        if (code) {
235,002,468✔
UNCOV
317
          pTaskInfo->code = code;
×
UNCOV
318
          tableListDestroy(pTableListInfo);
×
UNCOV
319
          return code;
×
320
        }
321
      }
322

323
      if (pTableScanNode->scan.node.dynamicOp) {
235,203,314✔
324
        pTaskInfo->dynamicTask = true;
4,629,367✔
325
        pTableListInfo->idInfo.suid = pTableScanNode->scan.suid;
4,630,814✔
326
        pTableListInfo->idInfo.tableType = pTableScanNode->scan.tableType;
4,628,450✔
327
      } else {
328
        code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
230,608,574✔
329
                                       pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo, NULL);
330
        if (code) {
230,914,007✔
331
          pTaskInfo->code = code;
888✔
332
          tableListDestroy(pTableListInfo);
888✔
333
          qError("failed to createScanTableListInfo, code:%s, %s", tstrerror(code), idstr);
888✔
334
          return code;
888✔
335
        }
336
      }
337

338
      code = createTableScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
235,543,327✔
339
      if (NULL == pOperator || code != 0) {
235,328,264✔
340
        pTaskInfo->code = code;
110,682✔
341
        tableListDestroy(pTableListInfo);
22,388✔
342
        return code;
22,388✔
343
      }
344

345
      STableScanInfo* pScanInfo = pOperator->info;
235,217,582✔
346
      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
235,336,037✔
347
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
151,168,486✔
348
      STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
24,782,253✔
349
      STableListInfo*           pTableListInfo = tableListCreate();
24,782,253✔
350
      if (!pTableListInfo) {
24,807,492✔
351
        pTaskInfo->code = terrno;
×
352
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
353
        return terrno;
×
354
      }
355

356
      code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle, pTableListInfo,
24,807,492✔
357
                                     pTagCond, pTagIndexCond, pTaskInfo, NULL);
358
      if (code) {
24,802,591✔
359
        pTaskInfo->code = code;
×
360
        tableListDestroy(pTableListInfo);
×
361
        qError("failed to createScanTableListInfo, code:%s", tstrerror(code));
×
362
        return code;
×
363
      }
364

365
      code = initQueriedTableSchemaInfo(pHandle, &pTableScanNode->scan, dbname, pTaskInfo);
24,802,591✔
366
      if (code) {
24,714,430✔
367
        pTaskInfo->code = code;
×
368
        tableListDestroy(pTableListInfo);
×
369
        return code;
×
370
      }
371

372
      code = createTableMergeScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
24,714,430✔
373
      if (NULL == pOperator || code != 0) {
24,760,524✔
374
        pTaskInfo->code = code;
63,671✔
375
        tableListDestroy(pTableListInfo);
×
376
        return code;
×
377
      }
378

379
      STableMergeScanInfo* pScanInfo = pOperator->info;
24,696,853✔
380
      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
24,770,058✔
381
    } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
126,386,233✔
382
      code = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode,
102,745,130✔
383
                                        pTaskInfo, &pOperator);
384
    } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
23,641,103✔
385
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
308,791✔
386
      STableListInfo*      pTableListInfo = tableListCreate();
308,791✔
387
      if (!pTableListInfo) {
308,791✔
388
        pTaskInfo->code = terrno;
×
389
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
390
        return terrno;
×
391
      }
392

393
      if (pHandle->vnode && (pTaskInfo->pSubplan->pVTables == NULL)) {
308,791✔
394
        code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
308,791✔
395
                                       pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo, NULL);
396
        if (code) {
308,791✔
397
          pTaskInfo->code = code;
×
398
          tableListDestroy(pTableListInfo);
×
399
          qError("failed to createScanTableListInfo, code:%s", tstrerror(code));
×
400
          return code;
×
401
        }
402
      }
403

404
      code = createTmqScanOperatorInfo(pHandle, pTableScanNode, pTagCond, pTableListInfo, pTaskInfo, &pOperator);
308,791✔
405
      if (code) {
308,791✔
406
        pTaskInfo->code = code;
×
407
        tableListDestroy(pTableListInfo);
×
408
        return code;
×
409
      }
410
    } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
23,332,312✔
411
      SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode;
10,330,987✔
412
      if (pSysScanPhyNode->scan.virtualStableScan) {
10,330,987✔
413
        STableListInfo*           pTableListInfo = tableListCreate();
400,996✔
414
        if (!pTableListInfo) {
400,996✔
415
          pTaskInfo->code = terrno;
×
416
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
417
          return terrno;
×
418
        }
419

420
        code = createScanTableListInfo((SScanPhysiNode*)pSysScanPhyNode, NULL, false, pHandle, pTableListInfo, pTagCond,
400,996✔
421
                                       pTagIndexCond, pTaskInfo, NULL);
422
        if (code != TSDB_CODE_SUCCESS) {
400,996✔
423
          pTaskInfo->code = code;
×
424
          tableListDestroy(pTableListInfo);
×
425
          return code;
×
426
        }
427

428
        code = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pTableListInfo, pUser, pTaskInfo, &pOperator);
400,996✔
429
      } else {
430
        code = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, NULL, pUser, pTaskInfo, &pOperator);
9,930,692✔
431
      }
432
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN == type) {
13,001,325✔
433
      STableCountScanPhysiNode* pTblCountScanNode = (STableCountScanPhysiNode*)pPhyNode;
24,914✔
434
      code = createTableCountScanOperatorInfo(pHandle, pTblCountScanNode, pTaskInfo, &pOperator);
24,914✔
435
    } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) {
12,976,411✔
436
      STagScanPhysiNode* pTagScanPhyNode = (STagScanPhysiNode*)pPhyNode;
8,372,222✔
437
      STableListInfo*    pTableListInfo = tableListCreate();
8,372,222✔
438
      if (!pTableListInfo) {
8,377,045✔
439
        pTaskInfo->code = terrno;
×
440
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
441
        return terrno;
×
442
      }
443

444
      code = initQueriedTableSchemaInfo(pHandle, &pTagScanPhyNode->scan, dbname, pTaskInfo);
8,377,045✔
445
      if (code != TSDB_CODE_SUCCESS) {
8,360,617✔
446
        pTaskInfo->code = code;
×
447
        tableListDestroy(pTableListInfo);
×
448
        return code;
×
449
      }
450

451
      if (!pTagScanPhyNode->onlyMetaCtbIdx) {
8,360,617✔
452
        code = createScanTableListInfo((SScanPhysiNode*)pTagScanPhyNode, NULL, false, pHandle, pTableListInfo, pTagCond,
3,058,088✔
453
                                       pTagIndexCond, pTaskInfo, NULL);
454
        if (code != TSDB_CODE_SUCCESS) {
3,058,088✔
455
          pTaskInfo->code = code;
×
456
          qError("failed to getTableList, code:%s", tstrerror(code));
×
457
          tableListDestroy(pTableListInfo);
×
458
          return code;
×
459
        }
460
      }
461

462
      if (pTagScanPhyNode->scan.node.dynamicOp) {
8,368,557✔
463
        pTaskInfo->dynamicTask = true;
124,564✔
464
        pTableListInfo->idInfo.suid = pTagScanPhyNode->scan.suid;
124,564✔
465
        pTableListInfo->idInfo.tableType = pTagScanPhyNode->scan.tableType;
124,564✔
466
      }
467

468
      code = createTagScanOperatorInfo(pHandle, pTagScanPhyNode, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo,
8,369,023✔
469
                                       &pOperator);
470
      if (code) {
8,372,800✔
471
        pTaskInfo->code = code;
×
472
        tableListDestroy(pTableListInfo);
×
473
        return code;
×
474
      }
475
    } else if (QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN == type) {
4,604,189✔
476
      SBlockDistScanPhysiNode* pBlockNode = (SBlockDistScanPhysiNode*)pPhyNode;
5,008✔
477
      STableListInfo*          pTableListInfo = tableListCreate();
5,008✔
478
      if (!pTableListInfo) {
5,008✔
479
        pTaskInfo->code = terrno;
×
480
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
481
        return terrno;
×
482
      }
483

484
      if (pBlockNode->tableType == TSDB_SUPER_TABLE) {
5,008✔
485
        SArray* pList = taosArrayInit(4, sizeof(uint64_t));
4,264✔
486
        code = pTaskInfo->storageAPI.metaFn.getChildTableList(pHandle->vnode, pBlockNode->uid, pList);
4,264✔
487
        if (code != TSDB_CODE_SUCCESS) {
4,264✔
488
          pTaskInfo->code = code;
×
489
          taosArrayDestroy(pList);
×
490
          tableListDestroy(pTableListInfo);
×
491
          return code;
×
492
        }
493

494
        size_t num = taosArrayGetSize(pList);
4,264✔
495
        for (int32_t i = 0; i < num; ++i) {
9,209✔
496
          uint64_t* id = taosArrayGet(pList, i);
4,945✔
497
          if (id == NULL) {
4,945✔
498
            continue;
×
499
          }
500

501
          code = tableListAddTableInfo(pTableListInfo, *id, 0);
4,945✔
502
          if (code) {
4,945✔
503
            pTaskInfo->code = code;
×
504
            tableListDestroy(pTableListInfo);
×
505
            taosArrayDestroy(pList);
×
506
            return code;
×
507
          }
508
        }
509

510
        taosArrayDestroy(pList);
4,264✔
511
      } else {  // Create group with only one table
512
        code = tableListAddTableInfo(pTableListInfo, pBlockNode->uid, 0);
744✔
513
        if (code) {
744✔
514
          pTaskInfo->code = code;
×
515
          tableListDestroy(pTableListInfo);
×
516
          return code;
×
517
        }
518
      }
519

520
      code = createDataBlockInfoScanOperator(pHandle, pBlockNode, pTableListInfo, pTaskInfo, &pOperator);
5,008✔
521
      if (code) {
5,008✔
522
        pTaskInfo->code = code;
×
523
        tableListDestroy(pTableListInfo);
×
524
        return code;
×
525
      }
526
    } else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) {
4,599,181✔
527
      SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode;
669,748✔
528
      STableListInfo*        pTableListInfo = tableListCreate();
669,748✔
529
      if (!pTableListInfo) {
670,135✔
530
        pTaskInfo->code = terrno;
×
531
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
532
        return terrno;
×
533
      }
534

535
      code = createScanTableListInfo(&pScanNode->scan, pScanNode->pGroupTags, true, pHandle, pTableListInfo, pTagCond,
670,135✔
536
                                     pTagIndexCond, pTaskInfo, NULL);
537
      if (code != TSDB_CODE_SUCCESS) {
670,135✔
538
        pTaskInfo->code = code;
×
539
        tableListDestroy(pTableListInfo);
×
540
        return code;
×
541
      }
542

543
      code = initQueriedTableSchemaInfo(pHandle, &pScanNode->scan, dbname, pTaskInfo);
670,135✔
544
      if (code != TSDB_CODE_SUCCESS) {
668,974✔
545
        pTaskInfo->code = code;
×
546
        tableListDestroy(pTableListInfo);
×
547
        return code;
×
548
      }
549

550
      code = createCacherowsScanOperator(pScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
668,974✔
551
      if (code) {
670,135✔
552
        tableListDestroy(pTableListInfo);
×
553
        pTaskInfo->code = code;
×
554
        return code;
×
555
      }
556
    } else if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
3,929,433✔
557
      code = createProjectOperatorInfo(NULL, (SProjectPhysiNode*)pPhyNode, pTaskInfo, &pOperator);
3,925,712✔
558
    } else if (QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN == type) {
3,721✔
559
      // NOTE: this is an patch to fix the physical plan
560
      code = createVirtualTableMergeOperatorInfo(NULL, 0, (SVirtualScanPhysiNode*)pPhyNode, pTaskInfo, &pOperator);
5,256✔
561
    } else {
562
      code = TSDB_CODE_INVALID_PARA;
217✔
563
      pTaskInfo->code = code;
217✔
564
      return code;
×
565
    }
566

567
    if (pOperator != NULL) {  // todo moved away
386,279,805✔
568
      pOperator->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
386,454,496✔
569
    }
570

571
    *pOptrInfo = pOperator;
386,354,260✔
572
    return code;
386,470,242✔
573
  }
574

575
  size_t          size = LIST_LENGTH(pPhyNode->pChildren);
369,936,271✔
576
  SOperatorInfo** ops = taosMemoryCalloc(size, POINTER_BYTES);
370,227,084✔
577
  if (ops == NULL) {
370,032,537✔
578
    code = terrno;
×
579
    pTaskInfo->code = code;
×
580
    return code;
×
581
  }
582

583
  for (int32_t i = 0; i < size; ++i) {
774,249,761✔
584
    SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, i);
405,935,407✔
585
    code = createOperator(pChildNode, pTaskInfo, pHandle, pTagCond, pTagIndexCond, pUser, dbname, &ops[i], model);
405,785,576✔
586
    if (ops[i] == NULL || code != 0) {
405,763,836✔
587
      for (int32_t j = 0; j < i; ++j) {
1,358,791✔
588
        destroyOperator(ops[j]);
×
589
      }
590
      taosMemoryFree(ops);
1,358,791✔
591
      return code;
1,358,791✔
592
    }
593
  }
594

595
  SOperatorInfo* pOptr = NULL;
368,314,354✔
596
  if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
368,467,657✔
597
    code = createProjectOperatorInfo(ops[0], (SProjectPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
130,048,442✔
598
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_AGG == type) {
238,419,215✔
599
    SAggPhysiNode* pAggNode = (SAggPhysiNode*)pPhyNode;
122,079,626✔
600
    if (pAggNode->pGroupKeys != NULL) {
122,079,626✔
601
      code = createGroupOperatorInfo(ops[0], pAggNode, pTaskInfo, &pOptr);
46,896,632✔
602
    } else {
603
      code = createAggregateOperatorInfo(ops[0], pAggNode, pTaskInfo, &pOptr);
75,190,738✔
604
    }
605
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type) {
116,339,589✔
606
    SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
12,193,495✔
607
    code = createIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
12,193,495✔
608
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) {
104,146,094✔
609
    SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode;
1,228,568✔
610
    code = createMergeAlignedIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
1,228,568✔
611
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL == type) {
102,917,526✔
612
    SMergeIntervalPhysiNode* pIntervalPhyNode = (SMergeIntervalPhysiNode*)pPhyNode;
×
613
    code = createMergeIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
×
614
  } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
102,917,526✔
615
    code = createSortOperatorInfo(ops[0], (SSortPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
47,444,274✔
616
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT == type) {
55,473,252✔
617
    code = createGroupSortOperatorInfo(ops[0], (SGroupSortPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
18,495✔
618
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) {
55,454,757✔
619
    SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode;
14,397,365✔
620
    code = createMultiwayMergeOperatorInfo(ops, size, pMergePhyNode, pTaskInfo, &pOptr);
14,397,365✔
621
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION == type) {
41,057,392✔
622
    SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
2,485,130✔
623
    code = createSessionAggOperatorInfo(ops[0], pSessionNode, pTaskInfo, &pOptr);
2,485,130✔
624
  } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
38,572,262✔
625
    code = createPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
9,173,764✔
626
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE == type) {
29,398,498✔
627
    SStateWindowPhysiNode* pStateNode = (SStateWindowPhysiNode*)pPhyNode;
1,966,755✔
628
    code = createStatewindowOperatorInfo(ops[0], pStateNode, pTaskInfo, &pOptr);
1,966,755✔
629
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN == type) {
27,431,743✔
630
    code = createMergeJoinOperatorInfo(ops, size, (SSortMergeJoinPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
14,279,566✔
631
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_JOIN == type) {
13,152,177✔
632
    code = createHashJoinOperatorInfo(ops, size, (SHashJoinPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
1,098,034✔
633
  } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) {
12,054,143✔
634
    code = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
456,877✔
635
  } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) {
11,597,266✔
636
    code = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
2,768,063✔
637
  } else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) {
8,829,203✔
638
    code = createTimeSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
2,101,775✔
639
  } else if (QUERY_NODE_PHYSICAL_PLAN_FORECAST_FUNC == type) {
6,727,428✔
640
    code = createForecastOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
641
  } else if (QUERY_NODE_PHYSICAL_PLAN_ANALYSIS_FUNC == type) {
6,727,428✔
642
    code = createGenericAnalysisOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
643
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT == type) {
6,727,428✔
644
    code = createEventwindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
1,707,898✔
645
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_CACHE == type) {
5,019,530✔
646
    code = createGroupCacheOperatorInfo(ops, size, (SGroupCachePhysiNode*)pPhyNode, pTaskInfo, &pOptr);
1,097,310✔
647
  } else if (QUERY_NODE_PHYSICAL_PLAN_DYN_QUERY_CTRL == type) {
3,922,220✔
648
    code = createDynQueryCtrlOperatorInfo(ops, size, (SDynQueryCtrlPhysiNode*)pPhyNode, pTaskInfo, pHandle->pMsgCb, &pOptr);
1,222,045✔
649
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_COUNT == type) {
2,700,175✔
650
    code = createCountwindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
1,698,032✔
651
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ANOMALY == type) {
1,002,143✔
652
    code = createAnomalywindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
653
  } else if (QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN == type) {
1,002,143✔
654
    SVirtualScanPhysiNode* pVirtualTableScanNode = (SVirtualScanPhysiNode*)pPhyNode;
926,640✔
655
    // NOTE: this is an patch to fix the physical plan
656

657
    if (pVirtualTableScanNode->scan.node.pLimit != NULL) {
926,640✔
658
      pVirtualTableScanNode->groupSort = true;
×
659
    }
660
    code = createVirtualTableMergeOperatorInfo(ops, size, (SVirtualScanPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
926,640✔
661
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_EXTERNAL == type) {
75,503✔
662
    code = createExternalWindowOperator(ops[0], pPhyNode, pTaskInfo, &pOptr);
71,696✔
663
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_EXTERNAL == type) {
3,913✔
664
    code = createMergeAlignedExternalWindowOperator(ops[0], pPhyNode, pTaskInfo, &pOptr);
1,470✔
665
  } else {
666
    code = TSDB_CODE_INVALID_PARA;
2,443✔
667
    pTaskInfo->code = code;
2,443✔
668
    for (int32_t i = 0; i < size; ++i) {
×
669
      destroyOperator(ops[i]);
×
670
    }
671
    taosMemoryFree(ops);
×
672
    qError("invalid operator type %d", type);
×
673
    return code;
×
674
  }
675

676
  taosMemoryFree(ops);
368,739,002✔
677
  if (pOptr) {
368,541,833✔
678
    pOptr->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
360,971,098✔
679
  }
680

681
  *pOptrInfo = pOptr;
368,725,382✔
682
  return code;
368,710,512✔
683
}
684

685
void destroyOperator(SOperatorInfo* pOperator) {
763,555,266✔
686
  if (pOperator == NULL) {
763,555,266✔
687
    return;
7,554,196✔
688
  }
689

690
  freeResetOperatorParams(pOperator, OP_GET_PARAM, true);
756,001,070✔
691
  freeResetOperatorParams(pOperator, OP_NOTIFY_PARAM, true);
755,871,983✔
692

693
  if (pOperator->pDownstream != NULL) {
755,905,799✔
694
    for (int32_t i = 0; i < pOperator->numOfRealDownstream; ++i) {
758,603,705✔
695
      destroyOperator(pOperator->pDownstream[i]);
397,160,417✔
696
    }
697

698
    taosMemoryFreeClear(pOperator->pDownstream);
361,518,732✔
699
    pOperator->numOfDownstream = 0;
361,508,702✔
700
  }
701

702
  if (pOperator->fpSet.closeFn != NULL && pOperator->info != NULL) {
756,068,944✔
703
    pOperator->fpSet.closeFn(pOperator->info);
748,631,461✔
704
    pOperator->info = NULL;
748,686,512✔
705
  }
706

707
  cleanupExprSupp(&pOperator->exprSupp);
756,196,803✔
708
  taosMemoryFreeClear(pOperator);
756,218,025✔
709
}
710

711
void destroyOperatorAndDownstreams(SOperatorInfo* pOperator, SOperatorInfo** downstreams, int32_t num) {
7,529,030✔
712
  if (downstreams != NULL) {
7,529,030✔
713
    for (int i = 0; i < num; i++) {
15,060,607✔
714
      destroyOperator(downstreams[i]);
7,529,313✔
715
    }
716
  }
717

718
  if (pOperator != NULL) {
7,530,445✔
719
    pOperator->info = NULL;
7,531,011✔
720
    if (pOperator->pDownstream != NULL) {
7,531,203✔
721
      taosMemoryFreeClear(pOperator->pDownstream);
×
722
      pOperator->pDownstream = NULL;
×
723
    }
724
    destroyOperator(pOperator);
7,530,728✔
725
  }
726
}
7,530,071✔
727

728
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
8,193,752✔
729
  SExplainExecInfo  execInfo = {0};
8,193,752✔
730
  SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
8,194,768✔
731
  if (pExplainInfo == NULL) {
8,194,768✔
732
    return terrno;
×
733
  }
734

735
  pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
8,194,768✔
736
  pExplainInfo->startupCost = operatorInfo->cost.openCost;
8,194,768✔
737
  pExplainInfo->totalCost = operatorInfo->cost.totalCost;
8,194,260✔
738
  pExplainInfo->verboseLen = 0;
8,194,260✔
739
  pExplainInfo->verboseInfo = NULL;
8,194,260✔
740

741
  if (operatorInfo->fpSet.getExplainFn) {
8,193,750✔
742
    int32_t code =
743
        operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
3,325,872✔
744
    if (code) {
3,324,346✔
745
      qError("%s operator getExplainFn failed, code:%s", GET_TASKID(operatorInfo->pTaskInfo), tstrerror(code));
×
746
      return code;
×
747
    }
748
  }
749

750
  int32_t code = 0;
8,193,244✔
751
  for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
12,686,468✔
752
    code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
4,492,716✔
753
    if (code != TSDB_CODE_SUCCESS) {
4,493,224✔
754
      //      taosMemoryFreeClear(*pRes);
755
      return code;
×
756
    }
757
  }
758

759
  return TSDB_CODE_SUCCESS;
8,194,260✔
760
}
761

762
int32_t mergeOperatorParams(SOperatorParam* pDst, SOperatorParam* pSrc) {
×
763
  if (pDst->opType != pSrc->opType) {
×
764
    qError("different optype %d:%d for merge operator params", pDst->opType, pSrc->opType);
×
765
    return TSDB_CODE_INVALID_PARA;
×
766
  }
767

768
  switch (pDst->opType) {
×
769
    case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE: {
×
770
      SExchangeOperatorParam* pDExc = pDst->value;
×
771
      SExchangeOperatorParam* pSExc = pSrc->value;
×
772
      if (!pDExc->multiParams) {
×
773
        if (pSExc->basic.vgId != pDExc->basic.vgId) {
×
774
          SExchangeOperatorBatchParam* pBatch = taosMemoryMalloc(sizeof(SExchangeOperatorBatchParam));
×
775
          if (NULL == pBatch) {
×
776
            return terrno;
×
777
          }
778

779
          pBatch->multiParams = true;
×
780
          pBatch->pBatchs = tSimpleHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));
×
781
          if (NULL == pBatch->pBatchs) {
×
782
            taosMemoryFree(pBatch);
×
783
            return terrno;
×
784
          }
785

786
          tSimpleHashSetFreeFp(pBatch->pBatchs, freeExchangeGetBasicOperatorParam);
×
787

788
          int32_t code = tSimpleHashPut(pBatch->pBatchs, &pDExc->basic.vgId, sizeof(pDExc->basic.vgId), &pDExc->basic,
×
789
                                        sizeof(pDExc->basic));
790
          if (code) {
×
791
            return code;
×
792
          }
793

794
          code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic,
×
795
                                sizeof(pSExc->basic));
796
          if (code) {
×
797
            return code;
×
798
          }
799

800
          taosMemoryFree(pDst->value);
×
801
          pDst->value = pBatch;
×
802
        } else {
803
          void* p = taosArrayAddAll(pDExc->basic.uidList, pSExc->basic.uidList);
×
804
          if (p == NULL) {
×
805
            return terrno;
×
806
          }
807
        }
808
      } else {
809
        SExchangeOperatorBatchParam* pBatch = pDst->value;
×
810
        SExchangeOperatorBasicParam* pBasic =
811
            tSimpleHashGet(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId));
×
812
        if (pBasic) {
×
813
          void* p = taosArrayAddAll(pBasic->uidList, pSExc->basic.uidList);
×
814
          if (p == NULL) {
×
815
            return terrno;
×
816
          }
817
        } else {
818
          int32_t code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic,
×
819
                                        sizeof(pSExc->basic));
820
          if (code) {
×
821
            return code;
×
822
          }
823
        }
824
      }
825
      break;
×
826
    }
827
    default:
×
828
      qError("invalid optype %d for merge operator params", pDst->opType);
×
829
      return TSDB_CODE_INVALID_PARA;
×
830
  }
831

832
  return TSDB_CODE_SUCCESS;
×
833
}
834

835
int32_t setOperatorParams(struct SOperatorInfo* pOperator, SOperatorParam* pInput, SOperatorParamType type) {
7,248,216✔
836
  SOperatorParam**  ppParam = NULL;
7,248,216✔
837
  SOperatorParam*** pppDownstramParam = NULL;
7,248,216✔
838
  switch (type) {
7,248,216✔
839
    case OP_GET_PARAM:
7,248,216✔
840
      ppParam = &pOperator->pOperatorGetParam;
7,248,216✔
841
      pppDownstramParam = &pOperator->pDownstreamGetParams;
7,248,216✔
842
      break;
7,248,216✔
843
    case OP_NOTIFY_PARAM:
×
844
      ppParam = &pOperator->pOperatorNotifyParam;
×
845
      pppDownstramParam = &pOperator->pDownstreamNotifyParams;
×
846
      break;
×
847
    default:
×
848
      return TSDB_CODE_INVALID_PARA;
×
849
  }
850

851
  freeResetOperatorParams(pOperator, type, false);
7,248,216✔
852

853
  if (NULL == pInput) {
7,248,216✔
854
    return TSDB_CODE_SUCCESS;
×
855
  }
856

857
  *ppParam = (pInput->opType == pOperator->operatorType) ? pInput : NULL;
7,248,216✔
858

859
  if (NULL == *pppDownstramParam) {
7,248,216✔
860
    *pppDownstramParam = taosMemoryCalloc(pOperator->numOfDownstream, POINTER_BYTES);
1,015,393✔
861
    if (NULL == *pppDownstramParam) {
1,015,393✔
862
      return terrno;
×
863
    }
864
  }
865

866
  if (NULL == *ppParam) {
7,248,216✔
867
    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
×
868
      (*pppDownstramParam)[i] = pInput;
×
869
    }
870
    return TSDB_CODE_SUCCESS;
×
871
  }
872

873
  memset(*pppDownstramParam, 0, pOperator->numOfDownstream * POINTER_BYTES);
7,248,216✔
874

875
  int32_t childrenNum = taosArrayGetSize((*ppParam)->pChildren);
7,248,216✔
876
  if (childrenNum <= 0) {
7,248,216✔
877
    return TSDB_CODE_SUCCESS;
3,709,540✔
878
  }
879

880
  for (int32_t i = 0; i < childrenNum; ++i) {
10,616,028✔
881
    SOperatorParam* pChild = *(SOperatorParam**)taosArrayGet((*ppParam)->pChildren, i);
7,077,352✔
882
    if (pChild == NULL) {
7,077,352✔
883
      return terrno;
×
884
    }
885

886
    if ((*pppDownstramParam)[pChild->downstreamIdx]) {
7,077,352✔
887
      int32_t code = mergeOperatorParams((*pppDownstramParam)[pChild->downstreamIdx], pChild);
×
888
      if (code) {
×
889
        return code;
×
890
      }
891
    } else {
892
      (*pppDownstramParam)[pChild->downstreamIdx] = pChild;
7,077,352✔
893
    }
894
  }
895

896
  taosArrayDestroy((*ppParam)->pChildren);
3,538,676✔
897
  (*ppParam)->pChildren = NULL;
3,538,676✔
898

899
  return TSDB_CODE_SUCCESS;
3,538,676✔
900
}
901

902
SSDataBlock* getNextBlockFromDownstream(struct SOperatorInfo* pOperator, int32_t idx) {
1,010,486,602✔
903
  SSDataBlock* p = NULL;
1,010,486,602✔
904
  int32_t      code = getNextBlockFromDownstreamImpl(pOperator, idx, true, &p);
1,010,599,336✔
905
  if (code == TSDB_CODE_SUCCESS) {
1,008,423,210✔
906
    code = blockDataCheck(p);
1,008,512,002✔
907
    if (code != TSDB_CODE_SUCCESS) {
1,008,548,395✔
908
      qError("blockDataCheck failed, code:%s", tstrerror(code));
×
909
    }
910
  }
911
  return (code == 0) ? p : NULL;
1,008,534,053✔
912
}
913

914
SSDataBlock* getNextBlockFromDownstreamRemain(struct SOperatorInfo* pOperator, int32_t idx) {
387,111,627✔
915
  SSDataBlock* p = NULL;
387,111,627✔
916
  int32_t      code = getNextBlockFromDownstreamImpl(pOperator, idx, false, &p);
387,142,993✔
917
  if (code == TSDB_CODE_SUCCESS) {
385,954,286✔
918
    code = blockDataCheck(p);
385,976,439✔
919
    if (code != TSDB_CODE_SUCCESS) {
385,989,815✔
920
      qError("blockDataCheck failed, code:%s", tstrerror(code));
×
921
    }
922
  }
923
  return (code == 0)? p:NULL;
385,987,431✔
924
}
925

926
int32_t optrDefaultGetNextExtFn(struct SOperatorInfo* pOperator, SOperatorParam* pParam, SSDataBlock** pRes) {
7,248,216✔
927
  QRY_PARAM_CHECK(pRes);
7,248,216✔
928

929
  int32_t lino = 0;
7,248,216✔
930
  int32_t code = setOperatorParams(pOperator, pParam, OP_GET_PARAM);
7,248,216✔
931
  QUERY_CHECK_CODE(code, lino, _end);
7,248,216✔
932
  code = pOperator->fpSet.getNextFn(pOperator, pRes);
7,248,216✔
933
  QUERY_CHECK_CODE(code, lino, _end);
7,246,028✔
934

935
_end:
7,246,028✔
936
  if (code != TSDB_CODE_SUCCESS) {
7,246,028✔
937
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
938
    pOperator->pTaskInfo->code = code;
×
939
    T_LONG_JMP(pOperator->pTaskInfo->env, code);
×
940
  }
941

942
  return code;
7,246,028✔
943
}
944

945
int32_t optrDefaultNotifyFn(struct SOperatorInfo* pOperator, SOperatorParam* pParam) {
×
946
  int32_t code = setOperatorParams(pOperator, pParam, OP_NOTIFY_PARAM);
×
947
  if (TSDB_CODE_SUCCESS == code && pOperator->fpSet.notifyFn && pOperator->pOperatorNotifyParam) {
×
948
    code = pOperator->fpSet.notifyFn(pOperator, pOperator->pOperatorNotifyParam);
×
949
  }
950
  if (TSDB_CODE_SUCCESS == code) {
×
951
    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
×
952
      if (pOperator->pDownstreamNotifyParams[i]) {
×
953
        code = optrDefaultNotifyFn(pOperator->pDownstream[i], pOperator->pDownstreamNotifyParams[i]);
×
954
        if (TSDB_CODE_SUCCESS != code) {
×
955
          break;
×
956
        }
957
        pOperator->pDownstreamNotifyParams[i] = NULL;
×
958
      }
959
    }
960
  }
961
  if (TSDB_CODE_SUCCESS != code) {
×
962
    pOperator->pTaskInfo->code = code;
×
963
    T_LONG_JMP(pOperator->pTaskInfo->env, pOperator->pTaskInfo->code);
×
964
  }
965

966
  return code;
×
967
}
968

969
int16_t getOperatorResultBlockId(struct SOperatorInfo* pOperator, int32_t idx) {
30,813,000✔
970
  if (pOperator->transparent) {
30,813,000✔
971
    return getOperatorResultBlockId(pOperator->pDownstream[idx], 0);
2,194,620✔
972
  }
973
  return pOperator->resultDataBlockId;
28,618,843✔
974
}
975

976
void resetOperatorState(SOperatorInfo *pOper) {
×
977
  pOper->status = OP_NOT_OPENED;
×
978
}
×
979

980
void resetBasicOperatorState(SOptrBasicInfo *pBasicInfo) {
12,562,670✔
981
  if (pBasicInfo->pRes) blockDataCleanup(pBasicInfo->pRes);
12,562,670✔
982
  initResultRowInfo(&pBasicInfo->resultRowInfo);
12,562,172✔
983
}
12,561,574✔
984

985
int32_t resetAggSup(SExprSupp* pExprSupp, SAggSupporter* pSup, SExecTaskInfo* pTaskInfo,
11,073,338✔
986
                    SNodeList* pNodeList, SNodeList* pGroupKeys, size_t keyBufSize, const char* pKey, void* pState,
987
                    SFunctionStateStore* pStore) {
988
  int32_t    code = 0, lino = 0, num = 0;
11,073,338✔
989
  SExprInfo* pExprInfo = NULL;
11,073,537✔
990
  cleanupAggSup(pSup);
11,073,537✔
991
  cleanupExprSuppWithoutFilter(pExprSupp);
11,073,300✔
992
  code = createExprInfo(pNodeList, pGroupKeys, &pExprInfo, &num);
11,072,809✔
993
  QUERY_CHECK_CODE(code, lino, _error);
11,073,736✔
994
  code = initAggSup(pExprSupp, pSup, pExprInfo, num, keyBufSize, pKey, pState, pStore);
11,073,736✔
995
  QUERY_CHECK_CODE(code, lino, _error);
11,072,607✔
996
  return code;
11,072,607✔
997
_error:
×
998
  if (code != TSDB_CODE_SUCCESS) {
×
999
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
1000
    pTaskInfo->code = code;
×
1001
  }
1002
  return code;
×
1003
}
1004

1005
int32_t resetExprSupp(SExprSupp* pExprSupp, SExecTaskInfo* pTaskInfo, SNodeList* pNodeList,
4,250,586✔
1006
                      SNodeList* pGroupKeys, SFunctionStateStore* pStore) {
1007
  int32_t code = 0, lino = 0, num = 0;
4,250,586✔
1008
  SExprInfo* pExprInfo = NULL;
4,250,586✔
1009
  cleanupExprSuppWithoutFilter(pExprSupp);
4,250,353✔
1010
  code = createExprInfo(pNodeList, pGroupKeys, &pExprInfo, &num);
4,250,785✔
1011
  QUERY_CHECK_CODE(code, lino, _error);
4,250,586✔
1012
  code = initExprSupp(pExprSupp, pExprInfo, num, pStore);
4,250,586✔
1013
  QUERY_CHECK_CODE(code, lino, _error);
4,250,785✔
1014
  return code;
4,250,785✔
1015
_error:
×
1016
  if (code != TSDB_CODE_SUCCESS) {
×
1017
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
1018
    pTaskInfo->code = code;
×
1019
  }
1020
  return code;
×
1021
}
1022

1023
int32_t copyColumnsValue(SNodeList* pNodeList, uint64_t targetBlkId, SSDataBlock* pDst, SSDataBlock* pSrc, int32_t totalRows) {
61,100✔
1024
  bool    isNull = (NULL == pSrc || pSrc->info.rows <= 0);
61,100✔
1025
  size_t  numOfCols = LIST_LENGTH(pNodeList);
61,100✔
1026
  int64_t numOfRows = isNull ? 0 : pSrc->info.rows;
61,100✔
1027
  int32_t code = 0;
61,100✔
1028
  int32_t lino = 0;
61,100✔
1029

1030
  for (int32_t i = 0; i < numOfCols; ++i) {
540,894✔
1031
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);
479,794✔
1032
    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN && ((SColumnNode*)pNode->pExpr)->dataBlockId == targetBlkId) {
479,794✔
1033
      SColumnInfoData* pDstCol = taosArrayGet(pDst->pDataBlock, pNode->slotId);
239,897✔
1034
      QUERY_CHECK_NULL(pDstCol, code, lino, _return, terrno)
239,897✔
1035

1036
      if (isNull) {
239,897✔
1037
        colDataSetNNULL(pDstCol, 0, totalRows);
×
1038
      } else {
1039
        SColumnInfoData* pSrcCol = taosArrayGet(pSrc->pDataBlock, ((SColumnNode*)pNode->pExpr)->slotId);
239,897✔
1040
        QUERY_CHECK_NULL(pSrcCol, code, lino, _return, terrno)
239,897✔
1041

1042
        code = colDataAssign(pDstCol, pSrcCol, pSrc->info.rows, &pDst->info);
239,897✔
1043

1044
        QUERY_CHECK_CODE(code, lino, _return);
239,897✔
1045
        if (pSrc->info.rows < totalRows) {
239,897✔
1046
          colDataSetNNULL(pDstCol, pSrc->info.rows, totalRows - pSrc->info.rows);
×
1047
        }
1048
      }
1049
    }
1050
  }
1051

1052
  return code;
61,100✔
1053
_return:
×
1054
  qError("failed to copy columns value, line:%d code:%s", lino, tstrerror(code));
×
1055
  return code;
×
1056
}
1057

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