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

taosdata / TDengine / #3848

15 Apr 2025 08:24AM UTC coverage: 62.61% (-0.002%) from 62.612%
#3848

push

travis-ci

web-flow
fix(dataformat): fix single column row copying (#30788)

154522 of 315314 branches covered (49.01%)

Branch coverage included in aggregate %.

34 of 47 new or added lines in 1 file covered. (72.34%)

540 existing lines in 131 files now uncovered.

240217 of 315158 relevant lines covered (76.22%)

19679321.97 hits per line

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

51.96
/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 "os.h"
19
#include "tname.h"
20

21
#include "tglobal.h"
22

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

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

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

48
  return fpSet;
20,193,976✔
49
}
50

51
void setOperatorStreamStateFn(SOperatorInfo* pOperator, __optr_state_fn_t relaseFn, __optr_state_fn_t reloadFn) {
1,842,398✔
52
  pOperator->fpSet.releaseStreamStateFn = relaseFn;
1,842,398✔
53
  pOperator->fpSet.reloadStreamStateFn = reloadFn;
1,842,398✔
54
}
1,842,398✔
55

56
int32_t optrDummyOpenFn(SOperatorInfo* pOperator) {
9,733,191✔
57
  OPTR_SET_OPENED(pOperator);
9,733,191✔
58
  pOperator->cost.openCost = 0;
9,733,191✔
59
  return TSDB_CODE_SUCCESS;
9,733,191✔
60
}
61

62
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num) {
10,074,664✔
63
  p->pDownstream = taosMemoryCalloc(1, num * POINTER_BYTES);
10,074,664!
64
  if (p->pDownstream == NULL) {
10,085,366!
65
    return terrno;
×
66
  }
67

68
  memcpy(p->pDownstream, pDownstream, num * POINTER_BYTES);
10,085,366✔
69
  p->numOfDownstream = num;
10,085,366✔
70
  p->numOfRealDownstream = num;
10,085,366✔
71
  return TSDB_CODE_SUCCESS;
10,085,366✔
72
}
73

74
void setOperatorCompleted(SOperatorInfo* pOperator) {
18,012,728✔
75
  pOperator->status = OP_EXEC_DONE;
18,012,728✔
76
  pOperator->cost.totalCost = (taosGetTimestampUs() - pOperator->pTaskInfo->cost.start) / 1000.0;
18,014,606✔
77
  setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
18,014,606✔
78
}
18,014,619✔
79

80
void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
20,188,175✔
81
                     void* pInfo, SExecTaskInfo* pTaskInfo) {
82
  pOperator->name = (char*)name;
20,188,175✔
83
  pOperator->operatorType = type;
20,188,175✔
84
  pOperator->blocking = blocking;
20,188,175✔
85
  pOperator->status = status;
20,188,175✔
86
  pOperator->info = pInfo;
20,188,175✔
87
  pOperator->pTaskInfo = pTaskInfo;
20,188,175✔
88
}
20,188,175✔
89

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

99
static int64_t getQuerySupportBufSize(size_t numOfTables) {
×
100
  size_t s1 = sizeof(STableQueryInfo);
×
101
  //  size_t s3 = sizeof(STableCheckInfo);  buffer consumption in tsdb
102
  return (int64_t)(s1 * 1.5 * numOfTables);
×
103
}
104

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

123
  // disable query processing if the value of tsQueryBufferSize is zero.
124
  return TSDB_CODE_QRY_NOT_ENOUGH_BUFFER;
×
125
}
126

127
void releaseQueryBuf(size_t numOfTables) {
×
128
  if (tsQueryBufferSizeBytes < 0) {
×
129
    return;
×
130
  }
131

132
  int64_t t = getQuerySupportBufSize(numOfTables);
×
133

134
  // restore value is not enough buffer available
135
  (void)atomic_add_fetch_64(&tsQueryBufferSizeBytes, t);
×
136
}
137

138
typedef enum {
139
  OPTR_FN_RET_CONTINUE = 0x1,
140
  OPTR_FN_RET_ABORT = 0x2,
141
} ERetType;
142

143
typedef struct STraverParam {
144
  void*   pRet;
145
  int32_t code;
146
  void*   pParam;
147
} STraverParam;
148

149
// iterate the operator tree helper
150
typedef ERetType (*optr_fn_t)(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdstr);
151

152
void traverseOperatorTree(SOperatorInfo* pOperator, optr_fn_t fn, STraverParam* pParam, const char* id) {
712,604✔
153
  if (pOperator == NULL) {
712,604!
154
    return;
×
155
  }
156

157
  ERetType ret = fn(pOperator, pParam, id);
712,604✔
158
  if (ret == OPTR_FN_RET_ABORT || pParam->code != TSDB_CODE_SUCCESS) {
712,605✔
159
    return;
337,549✔
160
  }
161

162
  for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
737,766✔
163
    traverseOperatorTree(pOperator->pDownstream[i], fn, pParam, id);
362,703✔
164
    if (pParam->code != 0) {
362,710!
165
      break;
×
166
    }
167
  }
168
}
169

170
ERetType extractOperatorInfo(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
667,884✔
171
  STraverParam* p = pParam;
667,884✔
172
  if (pOperator->operatorType == *(int32_t*)p->pParam) {
667,884✔
173
    p->pRet = pOperator;
329,515✔
174
    return OPTR_FN_RET_ABORT;
329,515✔
175
  } else {
176
    return OPTR_FN_RET_CONTINUE;
338,369✔
177
  }
178
}
179

180
// QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN
181
int32_t extractOperatorInTree(SOperatorInfo* pOperator, int32_t type, const char* id, SOperatorInfo** pOptrInfo) {
329,519✔
182
  QRY_PARAM_CHECK(pOptrInfo);
329,519!
183

184
  if (pOperator == NULL) {
329,519✔
185
    qError("invalid operator, failed to find tableScanOperator %s", id);
5!
186
    return TSDB_CODE_PAR_INTERNAL_ERROR;
5✔
187
  }
188

189
  STraverParam p = {.pParam = &type, .pRet = NULL};
329,514✔
190
  traverseOperatorTree(pOperator, extractOperatorInfo, &p, id);
329,514✔
191
  if (p.code == 0) {
329,514!
192
    *pOptrInfo = p.pRet;
329,515✔
193
  }
194
  return p.code;
329,514✔
195
}
196

197
typedef struct SExtScanInfo {
198
  int32_t order;
199
  int32_t scanFlag;
200
  int32_t inheritUsOrder;
201
} SExtScanInfo;
202

203
static ERetType extractScanInfo(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
×
204
  int32_t       type = pOperator->operatorType;
×
205
  SExtScanInfo* pInfo = pParam->pParam;
×
206

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

234
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag, bool inheritUsOrder) {
×
235
  SExtScanInfo info = {.inheritUsOrder = inheritUsOrder, .order = *order};
×
236
  STraverParam p = {.pParam = &info};
×
237

238
  traverseOperatorTree(pOperator, extractScanInfo, &p, NULL);
×
239
  *order = info.order;
×
240
  *scanFlag = info.scanFlag;
×
241

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

251
static ERetType doStopDataReader(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
44,719✔
252
  SStorageAPI* pAPI = pParam->pParam;
44,719✔
253
  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
44,719✔
254
    STableScanInfo* pInfo = pOperator->info;
780✔
255

256
    if (pInfo->base.dataReader != NULL) {
780✔
257
      pAPI->tsdReader.tsdReaderNotifyClosing(pInfo->base.dataReader);
770✔
258
    }
259
    return OPTR_FN_RET_ABORT;
780✔
260
  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
43,939✔
261
    SStreamScanInfo* pInfo = pOperator->info;
7,255✔
262

263
    if (pInfo->pTableScanOp != NULL) {
7,255✔
264
      STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
6,959✔
265
      if (pTableScanInfo != NULL && pTableScanInfo->base.dataReader != NULL) {
6,959!
266
        pAPI->tsdReader.tsdReaderNotifyClosing(pTableScanInfo->base.dataReader);
2✔
267
      }
268
    }
269

270
    return OPTR_FN_RET_ABORT;
7,255✔
271
  }
272

273
  return OPTR_FN_RET_CONTINUE;
36,684✔
274
}
275

276
int32_t stopTableScanOperator(SOperatorInfo* pOperator, const char* pIdStr, SStorageAPI* pAPI) {
20,411✔
277
  STraverParam p = {.pParam = pAPI};
20,411✔
278
  traverseOperatorTree(pOperator, doStopDataReader, &p, pIdStr);
20,411✔
279
  return p.code;
20,413✔
280
}
281

282
int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SNode* pTagCond,
20,189,037✔
283
                       SNode* pTagIndexCond, const char* pUser, const char* dbname, SOperatorInfo** pOptrInfo,
284
                       EOPTR_EXEC_MODEL model) {
285
  QRY_PARAM_CHECK(pOptrInfo);
20,189,037!
286

287
  int32_t     code = 0;
20,189,037✔
288
  int32_t     type = nodeType(pPhyNode);
20,189,037✔
289
  const char* idstr = GET_TASKID(pTaskInfo);
20,189,037✔
290

291
  if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
20,189,037!
292
    SOperatorInfo* pOperator = NULL;
10,106,321✔
293
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
10,106,321✔
294
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
3,006,563✔
295
      // NOTE: this is an patch to fix the physical plan
296
      // TODO remove it later
297
      if (pTableScanNode->scan.node.pLimit != NULL) {
3,006,563✔
298
        pTableScanNode->groupSort = true;
125,596✔
299
      }
300

301
      STableListInfo* pTableListInfo = tableListCreate();
3,006,563✔
302
      if (!pTableListInfo) {
3,007,215!
303
        pTaskInfo->code = terrno;
×
304
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
305
        return terrno;
×
306
      }
307

308
      // Since virtual stable scan use virtual super table's uid in scan operator, the origin table might be stored on
309
      // different vnode, so we should not get table schema for virtual stable scan.
310
      if (!pTableScanNode->scan.virtualStableScan) {
3,007,215✔
311
        code = initQueriedTableSchemaInfo(pHandle, &pTableScanNode->scan, dbname, pTaskInfo);
3,006,253✔
312
        if (code) {
3,004,016!
313
          pTaskInfo->code = code;
×
314
          tableListDestroy(pTableListInfo);
×
315
          return code;
116✔
316
        }
317
      }
318

319
      if (pTableScanNode->scan.node.dynamicOp) {
3,005,399✔
320
        pTaskInfo->dynamicTask = true;
44,124✔
321
        pTableListInfo->idInfo.suid = pTableScanNode->scan.suid;
44,124✔
322
        pTableListInfo->idInfo.tableType = pTableScanNode->scan.tableType;
44,124✔
323
      } else {
324
        code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
2,961,275✔
325
                                       pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
326
        if (code) {
2,963,724✔
327
          pTaskInfo->code = code;
24✔
328
          tableListDestroy(pTableListInfo);
24✔
329
          qError("failed to createScanTableListInfo, code:%s, %s", tstrerror(code), idstr);
24!
330
          return code;
24✔
331
        }
332
      }
333

334
      code = createTableScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
3,007,824✔
335
      if (NULL == pOperator || code != 0) {
3,002,367!
336
        pTaskInfo->code = code;
×
337
        tableListDestroy(pTableListInfo);
×
338
        return code;
220✔
339
      }
340

341
      STableScanInfo* pScanInfo = pOperator->info;
3,005,730✔
342
      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
3,005,730✔
343
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
7,099,758✔
344
      STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
2,051,092✔
345
      STableListInfo*           pTableListInfo = tableListCreate();
2,051,092✔
346
      if (!pTableListInfo) {
2,055,922!
347
        pTaskInfo->code = terrno;
×
348
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
349
        return terrno;
×
350
      }
351

352
      code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle, pTableListInfo,
2,055,922✔
353
                                     pTagCond, pTagIndexCond, pTaskInfo);
354
      if (code) {
2,054,949!
355
        pTaskInfo->code = code;
×
356
        tableListDestroy(pTableListInfo);
×
357
        qError("failed to createScanTableListInfo, code:%s", tstrerror(code));
×
358
        return code;
×
359
      }
360

361
      code = initQueriedTableSchemaInfo(pHandle, &pTableScanNode->scan, dbname, pTaskInfo);
2,054,949✔
362
      if (code) {
2,050,426!
363
        pTaskInfo->code = code;
×
364
        tableListDestroy(pTableListInfo);
×
365
        return code;
×
366
      }
367

368
      code = createTableMergeScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
2,050,426✔
369
      if (NULL == pOperator || code != 0) {
2,052,661!
370
        pTaskInfo->code = code;
×
371
        tableListDestroy(pTableListInfo);
×
372
        return code;
×
373
      }
374

375
      STableScanInfo* pScanInfo = pOperator->info;
2,054,416✔
376
      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
2,054,416✔
377
    } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
5,048,666✔
378
      code = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode,
3,742,791!
379
                                        pTaskInfo, &pOperator);
380
    } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
1,305,875✔
381
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
8,530✔
382
      STableListInfo*      pTableListInfo = tableListCreate();
8,530✔
383
      if (!pTableListInfo) {
8,530!
384
        pTaskInfo->code = terrno;
×
385
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
386
        return terrno;
×
387
      }
388

389
      if (pHandle->vnode && (pTaskInfo->pSubplan->pVTables == NULL)) {
8,530!
390
        code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
8,235✔
391
                                       pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
392
        if (code) {
8,235!
393
          pTaskInfo->code = code;
×
394
          tableListDestroy(pTableListInfo);
×
395
          qError("failed to createScanTableListInfo, code:%s", tstrerror(code));
×
396
          return code;
×
397
        }
398
      }
399

400
      code = createStreamScanOperatorInfo(pHandle, pTableScanNode, pTagCond, pTableListInfo, pTaskInfo, &pOperator);
8,530✔
401
      if (code) {
8,532!
402
        pTaskInfo->code = code;
×
403
        tableListDestroy(pTableListInfo);
×
404
        return code;
×
405
      }
406
    } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
1,297,345✔
407
      SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode;
1,243,355✔
408
      code = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pUser, pTaskInfo, &pOperator);
1,243,355✔
409
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN == type) {
53,990!
410
      STableCountScanPhysiNode* pTblCountScanNode = (STableCountScanPhysiNode*)pPhyNode;
×
411
      code = createTableCountScanOperatorInfo(pHandle, pTblCountScanNode, pTaskInfo, &pOperator);
×
412
    } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) {
53,990✔
413
      STagScanPhysiNode* pTagScanPhyNode = (STagScanPhysiNode*)pPhyNode;
47,893✔
414
      STableListInfo*    pTableListInfo = tableListCreate();
47,893✔
415
      if (!pTableListInfo) {
47,981!
416
        pTaskInfo->code = terrno;
×
417
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
418
        return terrno;
×
419
      }
420
      if (!pTagScanPhyNode->onlyMetaCtbIdx) {
47,981✔
421
        code = createScanTableListInfo((SScanPhysiNode*)pTagScanPhyNode, NULL, false, pHandle, pTableListInfo, pTagCond,
3,898✔
422
                                       pTagIndexCond, pTaskInfo);
423
        if (code != TSDB_CODE_SUCCESS) {
3,892!
424
          pTaskInfo->code = code;
×
425
          qError("failed to getTableList, code:%s", tstrerror(code));
×
426
          tableListDestroy(pTableListInfo);
×
427
          return code;
×
428
        }
429
      }
430
      code = createTagScanOperatorInfo(pHandle, pTagScanPhyNode, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo,
47,975✔
431
                                       &pOperator);
432
      if (code) {
47,995!
433
        pTaskInfo->code = code;
×
434
        tableListDestroy(pTableListInfo);
×
435
        return code;
×
436
      }
437
    } else if (QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN == type) {
6,097✔
438
      SBlockDistScanPhysiNode* pBlockNode = (SBlockDistScanPhysiNode*)pPhyNode;
3,247✔
439
      STableListInfo*          pTableListInfo = tableListCreate();
3,247✔
440
      if (!pTableListInfo) {
3,253✔
441
        pTaskInfo->code = terrno;
1✔
442
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
443
        return terrno;
×
444
      }
445

446
      if (pBlockNode->tableType == TSDB_SUPER_TABLE) {
3,252✔
447
        SArray* pList = taosArrayInit(4, sizeof(uint64_t));
3,249✔
448
        code = pTaskInfo->storageAPI.metaFn.getChildTableList(pHandle->vnode, pBlockNode->uid, pList);
3,247✔
449
        if (code != TSDB_CODE_SUCCESS) {
3,249!
450
          pTaskInfo->code = code;
×
451
          taosArrayDestroy(pList);
×
452
          tableListDestroy(pTableListInfo);
×
453
          return code;
×
454
        }
455

456
        size_t num = taosArrayGetSize(pList);
3,249✔
457
        for (int32_t i = 0; i < num; ++i) {
10,271✔
458
          uint64_t* id = taosArrayGet(pList, i);
7,021✔
459
          if (id == NULL) {
7,004!
460
            continue;
×
461
          }
462

463
          code = tableListAddTableInfo(pTableListInfo, *id, 0);
7,004✔
464
          if (code) {
7,025!
465
            pTaskInfo->code = code;
×
466
            tableListDestroy(pTableListInfo);
×
467
            taosArrayDestroy(pList);
×
468
            return code;
7,043✔
469
          }
470
        }
471

472
        taosArrayDestroy(pList);
3,250✔
473
      } else {  // Create group with only one table
474
        code = tableListAddTableInfo(pTableListInfo, pBlockNode->uid, 0);
3✔
475
        if (code) {
3!
476
          pTaskInfo->code = code;
×
477
          tableListDestroy(pTableListInfo);
×
478
          return code;
×
479
        }
480
      }
481

482
      code = createDataBlockInfoScanOperator(pHandle, pBlockNode, pTableListInfo, pTaskInfo, &pOperator);
3,251✔
483
      if (code) {
3,244!
484
        pTaskInfo->code = code;
×
485
        tableListDestroy(pTableListInfo);
×
486
        return code;
×
487
      }
488
    } else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) {
2,850!
489
      SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode;
7,341✔
490
      STableListInfo*        pTableListInfo = tableListCreate();
7,341✔
491
      if (!pTableListInfo) {
7,364!
492
        pTaskInfo->code = terrno;
×
493
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
494
        return terrno;
×
495
      }
496

497
      code = createScanTableListInfo(&pScanNode->scan, pScanNode->pGroupTags, true, pHandle, pTableListInfo, pTagCond,
7,364✔
498
                                     pTagIndexCond, pTaskInfo);
499
      if (code != TSDB_CODE_SUCCESS) {
7,361!
500
        pTaskInfo->code = code;
×
501
        tableListDestroy(pTableListInfo);
×
502
        return code;
×
503
      }
504

505
      code = initQueriedTableSchemaInfo(pHandle, &pScanNode->scan, dbname, pTaskInfo);
7,361✔
506
      if (code != TSDB_CODE_SUCCESS) {
7,356!
507
        pTaskInfo->code = code;
×
508
        tableListDestroy(pTableListInfo);
×
509
        return code;
×
510
      }
511

512
      code = createCacherowsScanOperator(pScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
7,356✔
513
      if (code) {
7,330!
514
        tableListDestroy(pTableListInfo);
×
515
        pTaskInfo->code = code;
×
516
        return code;
×
517
      }
518
    } else if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
×
519
      code = createProjectOperatorInfo(NULL, (SProjectPhysiNode*)pPhyNode, pTaskInfo, &pOperator);
1,657✔
520
    } else if (QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN == type && model != OPTR_EXEC_MODEL_STREAM) {
×
521
      code = createVirtualTableMergeOperatorInfo(NULL, pHandle, NULL, 0, (SVirtualScanPhysiNode*)pPhyNode, pTaskInfo, &pOperator);
×
522
    } else {
523
      code = TSDB_CODE_INVALID_PARA;
×
524
      pTaskInfo->code = code;
×
525
      return code;
×
526
    }
527

528
    if (pOperator != NULL) {  // todo moved away
10,111,803✔
529
      pOperator->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
10,109,328✔
530
    }
531

532
    *pOptrInfo = pOperator;
10,111,803✔
533
    return code;
10,111,803✔
534
  }
535

536
  size_t          size = LIST_LENGTH(pPhyNode->pChildren);
10,082,716✔
537
  SOperatorInfo** ops = taosMemoryCalloc(size, POINTER_BYTES);
10,082,716!
538
  if (ops == NULL) {
10,085,171!
539
    code = terrno;
×
540
    pTaskInfo->code = code;
×
541
    return code;
×
542
  }
543

544
  for (int32_t i = 0; i < size; ++i) {
21,755,658✔
545
    SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, i);
11,666,392✔
546
    code = createOperator(pChildNode, pTaskInfo, pHandle, pTagCond, pTagIndexCond, pUser, dbname, &ops[i], model);
11,671,018✔
547
    if (ops[i] == NULL || code != 0) {
11,670,603!
548
      for (int32_t j = 0; j < i; ++j) {
116!
549
        destroyOperator(ops[j]);
×
550
      }
551
      taosMemoryFree(ops);
116!
552
      return code;
116✔
553
    }
554
  }
555

556
  SOperatorInfo* pOptr = NULL;
10,089,266✔
557
  if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
10,089,266✔
558
    code = createProjectOperatorInfo(ops[0], (SProjectPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
1,825,641✔
559
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_AGG == type) {
8,263,625✔
560
    SAggPhysiNode* pAggNode = (SAggPhysiNode*)pPhyNode;
2,515,303✔
561
    if (pAggNode->pGroupKeys != NULL) {
2,515,303✔
562
      code = createGroupOperatorInfo(ops[0], pAggNode, pTaskInfo, &pOptr);
364,637✔
563
    } else {
564
      code = createAggregateOperatorInfo(ops[0], pAggNode, pTaskInfo, &pOptr);
2,150,666✔
565
    }
566
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type) {
5,748,322✔
567
    SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
1,514,119✔
568
    code = createIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
1,514,119✔
569
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type) {
4,234,203✔
570
    code = createStreamSingleIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
2,816✔
571
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL == type) {
4,231,387✔
572
    code = createStreamIntervalSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
107✔
573
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) {
4,231,280✔
574
    SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode;
555,870✔
575
    code = createMergeAlignedIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
555,870✔
576
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL == type) {
3,675,410!
577
    SMergeIntervalPhysiNode* pIntervalPhyNode = (SMergeIntervalPhysiNode*)pPhyNode;
×
578
    code = createMergeIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
×
579
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL == type) {
3,675,410✔
580
    int32_t children = 0;
663✔
581
    code = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children, pHandle, &pOptr);
663✔
582
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_INTERVAL == type) {
3,674,747✔
583
    code = createSemiIntervalSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
36✔
584
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL == type) {
3,674,711✔
585
    int32_t children = pHandle->numOfVgroups;
22✔
586
    code = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children, pHandle, &pOptr);
22✔
587
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL == type) {
3,674,689✔
588
    int32_t children = pHandle->numOfVgroups;
220✔
589
    code = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children, pHandle, &pOptr);
220✔
590
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_INTERVAL == type) {
3,674,469✔
591
    code = createFinalIntervalSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
18✔
592
  } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
3,674,451✔
593
    code = createSortOperatorInfo(ops[0], (SSortPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
912,369✔
594
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT == type) {
2,762,082✔
595
    code = createGroupSortOperatorInfo(ops[0], (SGroupSortPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
205✔
596
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) {
2,761,877✔
597
    SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode;
1,407,148✔
598
    code = createMultiwayMergeOperatorInfo(ops, size, pMergePhyNode, pTaskInfo, &pOptr);
1,407,148✔
599
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION == type) {
1,354,729✔
600
    SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
103,877✔
601
    code = createSessionAggOperatorInfo(ops[0], pSessionNode, pTaskInfo, &pOptr);
103,877✔
602
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION == type) {
1,250,852✔
603
    code = createStreamSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
617✔
604
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION == type) {
1,250,235✔
605
    int32_t children = 0;
125✔
606
    code = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children, pHandle, &pOptr);
125✔
607
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION == type) {
1,250,110✔
608
    int32_t children = pHandle->numOfVgroups;
38✔
609
    code = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children, pHandle, &pOptr);
38✔
610
  } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
1,250,072✔
611
    code = createPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
27,872✔
612
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION == type) {
1,222,200✔
613
    code = createStreamPartitionOperatorInfo(ops[0], (SStreamPartitionPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
1,220✔
614
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE == type) {
1,220,980✔
615
    SStateWinodwPhysiNode* pStateNode = (SStateWinodwPhysiNode*)pPhyNode;
99,792✔
616
    code = createStatewindowOperatorInfo(ops[0], pStateNode, pTaskInfo, &pOptr);
99,792✔
617
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE == type) {
1,121,188✔
618
    code = createStreamStateAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
350✔
619
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT == type) {
1,120,838✔
620
    code = createStreamEventAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
212✔
621
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN == type) {
1,120,626✔
622
    code = createMergeJoinOperatorInfo(ops, size, (SSortMergeJoinPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
100,074✔
623
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_JOIN == type) {
1,020,552✔
624
    code = createHashJoinOperatorInfo(ops, size, (SHashJoinPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
11,035✔
625
  } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) {
1,009,517✔
626
    code = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
494,262✔
627
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL == type) {
515,255✔
628
    code = createStreamFillOperatorInfo(ops[0], (SStreamFillPhysiNode*)pPhyNode, pTaskInfo, pHandle, &pOptr);
451✔
629
  } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) {
514,804✔
630
    code = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
117,503✔
631
  } else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) {
397,301✔
632
    code = createTimeSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
74,933✔
633
  } else if (QUERY_NODE_PHYSICAL_PLAN_FORECAST_FUNC == type) {
322,368!
634
    code = createForecastOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
635
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT == type) {
322,368✔
636
    code = createEventwindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
284,005✔
637
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_CACHE == type) {
38,363✔
638
    code = createGroupCacheOperatorInfo(ops, size, (SGroupCachePhysiNode*)pPhyNode, pTaskInfo, &pOptr);
11,035✔
639
  } else if (QUERY_NODE_PHYSICAL_PLAN_DYN_QUERY_CTRL == type) {
27,328✔
640
    code = createDynQueryCtrlOperatorInfo(ops, size, (SDynQueryCtrlPhysiNode*)pPhyNode, pTaskInfo, pHandle, &pOptr);
11,035✔
641
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT == type) {
16,293✔
642
    code = createStreamCountAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
314✔
643
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_COUNT == type) {
15,979✔
644
    code = createCountwindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
7,689✔
645
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC == type) {
8,290✔
646
    code = createStreamTimeSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
242✔
647
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ANOMALY == type) {
8,048!
648
    code = createAnomalywindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
649
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SESSION == type) {
8,048!
650
    code = createSessionNonblockOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
×
651
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_SESSION == type) {
8,048!
652
    code = createSemiSessionNonblockOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
×
653
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_SESSION == type) {
8,048!
654
    code = createFinalSessionNonblockOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
×
655
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_STATE == type) {
8,048!
656
    code = createStateNonblockOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
×
657
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_EVENT == type) {
8,048!
658
    code = createEventNonblockOperatorInfo(ops[0], pPhyNode, pTaskInfo, pHandle, &pOptr);
×
659
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_COUNT == type) {
8,048!
660
    //todo (liuyao) add
661
  } else if (QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN == type && model != OPTR_EXEC_MODEL_STREAM) {
×
662
    SVirtualScanPhysiNode* pVirtualTableScanNode = (SVirtualScanPhysiNode*)pPhyNode;
×
663
    // NOTE: this is an patch to fix the physical plan
664

665
    if (pVirtualTableScanNode->scan.node.pLimit != NULL) {
×
666
      pVirtualTableScanNode->groupSort = true;
×
667
    }
668

669
    STableListInfo* pTableListInfo = tableListCreate();
×
670
    if (!pTableListInfo) {
×
671
      pTaskInfo->code = terrno;
×
672
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
673
      return terrno;
×
674
    }
675

676
    code = createScanTableListInfo(&pVirtualTableScanNode->scan, pVirtualTableScanNode->pGroupTags, pVirtualTableScanNode->groupSort,
×
677
                                   pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
678
    if (code) {
×
679
      pTaskInfo->code = code;
×
680
      tableListDestroy(pTableListInfo);
×
681
      qError("failed to createScanTableListInfo, code:%s, %s", tstrerror(code), idstr);
×
682
      return code;
×
683
    }
684

685
    code = createVirtualTableMergeOperatorInfo(ops, pHandle, pTableListInfo, size, (SVirtualScanPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
×
686
  } else if (QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN == type && model == OPTR_EXEC_MODEL_STREAM) {
×
687
    SVirtualScanPhysiNode* pVirtualTableScanNode = (SVirtualScanPhysiNode*)pPhyNode;
×
688
    STableListInfo*        pTableListInfo = tableListCreate();
×
689
    if (!pTableListInfo) {
×
690
      pTaskInfo->code = terrno;
×
691
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
692
      return terrno;
×
693
    }
694

695
    code = createScanTableListInfo(&pVirtualTableScanNode->scan, pVirtualTableScanNode->pGroupTags,
×
696
                                   pVirtualTableScanNode->groupSort, pHandle, pTableListInfo, pTagCond, pTagIndexCond,
×
697
                                   pTaskInfo);
698
    if (code) {
×
699
      pTaskInfo->code = code;
×
700
      tableListDestroy(pTableListInfo);
×
701
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
702
      return code;
×
703
    }
704

705
    code = createStreamVtableMergeOperatorInfo(ops[0], pHandle, pVirtualTableScanNode, pTagCond, pTableListInfo, pTaskInfo, &pOptr);
×
706
  } else {
707
    code = TSDB_CODE_INVALID_PARA;
×
708
    pTaskInfo->code = code;
×
709
    for (int32_t i = 0; i < size; ++i) {
×
710
      destroyOperator(ops[i]);
×
711
    }
712
    taosMemoryFree(ops);
×
713
    qError("invalid operator type %d", type);
×
714
    return code;
×
715
  }
716

717
  taosMemoryFree(ops);
10,085,773✔
718
  if (pOptr) {
10,080,687✔
719
    pOptr->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
10,080,336✔
720
  }
721

722
  *pOptrInfo = pOptr;
10,080,687✔
723
  return code;
10,080,687✔
724
}
725

726
void destroyOperator(SOperatorInfo* pOperator) {
20,218,860✔
727
  if (pOperator == NULL) {
20,218,860✔
728
    return;
360✔
729
  }
730

731
  freeResetOperatorParams(pOperator, OP_GET_PARAM, true);
20,218,500✔
732
  freeResetOperatorParams(pOperator, OP_NOTIFY_PARAM, true);
20,218,476✔
733

734
  if (pOperator->pDownstream != NULL) {
20,219,268✔
735
    for (int32_t i = 0; i < pOperator->numOfRealDownstream; ++i) {
21,763,483✔
736
      destroyOperator(pOperator->pDownstream[i]);
11,674,822✔
737
    }
738

739
    taosMemoryFreeClear(pOperator->pDownstream);
10,088,661✔
740
    pOperator->numOfDownstream = 0;
10,088,420✔
741
  }
742

743
  if (pOperator->fpSet.closeFn != NULL && pOperator->info != NULL) {
20,219,704!
744
    pOperator->fpSet.closeFn(pOperator->info);
20,219,620✔
745
  }
746

747
  cleanupExprSupp(&pOperator->exprSupp);
20,220,229✔
748
  taosMemoryFreeClear(pOperator);
20,218,892!
749
}
750

751
void destroyOperatorAndDownstreams(SOperatorInfo* pOperator, SOperatorInfo** downstreams, int32_t num) {
×
752
  if (downstreams != NULL) {
×
753
    for (int i = 0; i < num; i++) {
×
754
      destroyOperator(downstreams[i]);
×
755
    }
756
  }
757

758
  if (pOperator != NULL) {
×
759
    pOperator->info = NULL;
×
760
    if (pOperator->pDownstream != NULL) {
×
761
      taosMemoryFreeClear(pOperator->pDownstream);
×
762
      pOperator->pDownstream = NULL;
×
763
    }
764
    destroyOperator(pOperator);
×
765
  }
766
}
×
767

768
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
1,355,332✔
769
  SExplainExecInfo  execInfo = {0};
1,355,332✔
770
  SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
1,355,339✔
771
  if (pExplainInfo == NULL) {
1,355,339!
772
    return terrno;
×
773
  }
774

775
  pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
1,355,339✔
776
  pExplainInfo->startupCost = operatorInfo->cost.openCost;
1,355,339✔
777
  pExplainInfo->totalCost = operatorInfo->cost.totalCost;
1,355,339✔
778
  pExplainInfo->verboseLen = 0;
1,355,339✔
779
  pExplainInfo->verboseInfo = NULL;
1,355,339✔
780

781
  if (operatorInfo->fpSet.getExplainFn) {
1,355,339✔
782
    int32_t code =
783
        operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
594,374✔
784
    if (code) {
594,386!
UNCOV
785
      qError("%s operator getExplainFn failed, code:%s", GET_TASKID(operatorInfo->pTaskInfo), tstrerror(code));
×
786
      return code;
×
787
    }
788
  }
789

790
  int32_t code = 0;
1,355,351✔
791
  for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
2,137,193✔
792
    code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
781,830✔
793
    if (code != TSDB_CODE_SUCCESS) {
781,842!
794
      //      taosMemoryFreeClear(*pRes);
795
      return code;
×
796
    }
797
  }
798

799
  return TSDB_CODE_SUCCESS;
1,355,363✔
800
}
801

802
int32_t mergeOperatorParams(SOperatorParam* pDst, SOperatorParam* pSrc) {
×
803
  if (pDst->opType != pSrc->opType) {
×
804
    qError("different optype %d:%d for merge operator params", pDst->opType, pSrc->opType);
×
805
    return TSDB_CODE_INVALID_PARA;
×
806
  }
807

808
  switch (pDst->opType) {
×
809
    case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE: {
×
810
      SExchangeOperatorParam* pDExc = pDst->value;
×
811
      SExchangeOperatorParam* pSExc = pSrc->value;
×
812
      if (!pDExc->multiParams) {
×
813
        if (pSExc->basic.vgId != pDExc->basic.vgId) {
×
814
          SExchangeOperatorBatchParam* pBatch = taosMemoryMalloc(sizeof(SExchangeOperatorBatchParam));
×
815
          if (NULL == pBatch) {
×
816
            return terrno;
×
817
          }
818

819
          pBatch->multiParams = true;
×
820
          pBatch->pBatchs = tSimpleHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));
×
821
          if (NULL == pBatch->pBatchs) {
×
822
            taosMemoryFree(pBatch);
×
823
            return terrno;
×
824
          }
825

826
          tSimpleHashSetFreeFp(pBatch->pBatchs, freeExchangeGetBasicOperatorParam);
×
827

828
          int32_t code = tSimpleHashPut(pBatch->pBatchs, &pDExc->basic.vgId, sizeof(pDExc->basic.vgId), &pDExc->basic,
×
829
                                        sizeof(pDExc->basic));
830
          if (code) {
×
831
            return code;
×
832
          }
833

834
          code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic,
×
835
                                sizeof(pSExc->basic));
836
          if (code) {
×
837
            return code;
×
838
          }
839

840
          taosMemoryFree(pDst->value);
×
841
          pDst->value = pBatch;
×
842
        } else {
843
          void* p = taosArrayAddAll(pDExc->basic.uidList, pSExc->basic.uidList);
×
844
          if (p == NULL) {
×
845
            return terrno;
×
846
          }
847
        }
848
      } else {
849
        SExchangeOperatorBatchParam* pBatch = pDst->value;
×
850
        SExchangeOperatorBasicParam* pBasic =
851
            tSimpleHashGet(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId));
×
852
        if (pBasic) {
×
853
          void* p = taosArrayAddAll(pBasic->uidList, pSExc->basic.uidList);
×
854
          if (p == NULL) {
×
855
            return terrno;
×
856
          }
857
        } else {
858
          int32_t code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic,
×
859
                                        sizeof(pSExc->basic));
860
          if (code) {
×
861
            return code;
×
862
          }
863
        }
864
      }
865
      break;
×
866
    }
867
    default:
×
868
      qError("invalid optype %d for merge operator params", pDst->opType);
×
869
      return TSDB_CODE_INVALID_PARA;
×
870
  }
871

872
  return TSDB_CODE_SUCCESS;
×
873
}
874

875
int32_t setOperatorParams(struct SOperatorInfo* pOperator, SOperatorParam* pInput, SOperatorParamType type) {
22,684✔
876
  SOperatorParam**  ppParam = NULL;
22,684✔
877
  SOperatorParam*** pppDownstramParam = NULL;
22,684✔
878
  switch (type) {
22,684!
879
    case OP_GET_PARAM:
22,684✔
880
      ppParam = &pOperator->pOperatorGetParam;
22,684✔
881
      pppDownstramParam = &pOperator->pDownstreamGetParams;
22,684✔
882
      break;
22,684✔
883
    case OP_NOTIFY_PARAM:
×
884
      ppParam = &pOperator->pOperatorNotifyParam;
×
885
      pppDownstramParam = &pOperator->pDownstreamNotifyParams;
×
886
      break;
×
887
    default:
×
888
      return TSDB_CODE_INVALID_PARA;
×
889
  }
890

891
  freeResetOperatorParams(pOperator, type, false);
22,684✔
892

893
  if (NULL == pInput) {
22,684!
894
    return TSDB_CODE_SUCCESS;
×
895
  }
896

897
  *ppParam = (pInput->opType == pOperator->operatorType) ? pInput : NULL;
22,684!
898

899
  if (NULL == *pppDownstramParam) {
22,684✔
900
    *pppDownstramParam = taosMemoryCalloc(pOperator->numOfDownstream, POINTER_BYTES);
1,910!
901
    if (NULL == *pppDownstramParam) {
1,910!
902
      return terrno;
×
903
    }
904
  }
905

906
  if (NULL == *ppParam) {
22,684!
907
    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
×
908
      (*pppDownstramParam)[i] = pInput;
×
909
    }
910
    return TSDB_CODE_SUCCESS;
×
911
  }
912

913
  memset(*pppDownstramParam, 0, pOperator->numOfDownstream * POINTER_BYTES);
22,684✔
914

915
  int32_t childrenNum = taosArrayGetSize((*ppParam)->pChildren);
22,684✔
916
  if (childrenNum <= 0) {
22,683✔
917
    return TSDB_CODE_SUCCESS;
1,639✔
918
  }
919

920
  for (int32_t i = 0; i < childrenNum; ++i) {
63,132✔
921
    SOperatorParam* pChild = *(SOperatorParam**)taosArrayGet((*ppParam)->pChildren, i);
42,088✔
922
    if (pChild == NULL) {
42,088!
923
      return terrno;
×
924
    }
925

926
    if ((*pppDownstramParam)[pChild->downstreamIdx]) {
42,088!
927
      int32_t code = mergeOperatorParams((*pppDownstramParam)[pChild->downstreamIdx], pChild);
×
928
      if (code) {
×
929
        return code;
×
930
      }
931
    } else {
932
      (*pppDownstramParam)[pChild->downstreamIdx] = pChild;
42,088✔
933
    }
934
  }
935

936
  taosArrayDestroy((*ppParam)->pChildren);
21,044✔
937
  (*ppParam)->pChildren = NULL;
21,044✔
938

939
  return TSDB_CODE_SUCCESS;
21,044✔
940
}
941

942
SSDataBlock* getNextBlockFromDownstream(struct SOperatorInfo* pOperator, int32_t idx) {
35,962,536✔
943
  SSDataBlock* p = NULL;
35,962,536✔
944
  int32_t      code = getNextBlockFromDownstreamImpl(pOperator, idx, true, &p);
35,962,536✔
945
  if (code == TSDB_CODE_SUCCESS) {
35,966,066!
946
    code = blockDataCheck(p);
35,967,496✔
947
    if (code != TSDB_CODE_SUCCESS) {
35,964,037✔
948
      qError("blockDataCheck failed, code:%s", tstrerror(code));
144!
949
    }
950
  }
951
  return (code == 0) ? p : NULL;
35,963,441✔
952
}
953

954
SSDataBlock* getNextBlockFromDownstreamRemain(struct SOperatorInfo* pOperator, int32_t idx) {
529,547✔
955
  SSDataBlock* p = NULL;
529,547✔
956
  int32_t      code = getNextBlockFromDownstreamImpl(pOperator, idx, false, &p);
529,547✔
957
  if (code == TSDB_CODE_SUCCESS) {
529,465!
958
    code = blockDataCheck(p);
529,512✔
959
    if (code != TSDB_CODE_SUCCESS) {
529,559!
960
      qError("blockDataCheck failed, code:%s", tstrerror(code));
×
961
    }
962
  }
963
  return (code == 0)? p:NULL;
529,552!
964
}
965

966
int32_t optrDefaultGetNextExtFn(struct SOperatorInfo* pOperator, SOperatorParam* pParam, SSDataBlock** pRes) {
22,684✔
967
  QRY_PARAM_CHECK(pRes);
22,684!
968

969
  int32_t lino = 0;
22,684✔
970
  int32_t code = setOperatorParams(pOperator, pParam, OP_GET_PARAM);
22,684✔
971
  QUERY_CHECK_CODE(code, lino, _end);
22,684!
972
  code = pOperator->fpSet.getNextFn(pOperator, pRes);
22,684✔
973
  QUERY_CHECK_CODE(code, lino, _end);
22,682!
974

975
_end:
22,682✔
976
  if (code != TSDB_CODE_SUCCESS) {
22,682!
977
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
978
    pOperator->pTaskInfo->code = code;
×
979
    T_LONG_JMP(pOperator->pTaskInfo->env, code);
×
980
  }
981

982
  return code;
22,682✔
983
}
984

985
int32_t optrDefaultNotifyFn(struct SOperatorInfo* pOperator, SOperatorParam* pParam) {
×
986
  int32_t code = setOperatorParams(pOperator, pParam, OP_NOTIFY_PARAM);
×
987
  if (TSDB_CODE_SUCCESS == code && pOperator->fpSet.notifyFn && pOperator->pOperatorNotifyParam) {
×
988
    code = pOperator->fpSet.notifyFn(pOperator, pOperator->pOperatorNotifyParam);
×
989
  }
990
  if (TSDB_CODE_SUCCESS == code) {
×
991
    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
×
992
      if (pOperator->pDownstreamNotifyParams[i]) {
×
993
        code = optrDefaultNotifyFn(pOperator->pDownstream[i], pOperator->pDownstreamNotifyParams[i]);
×
994
        if (TSDB_CODE_SUCCESS != code) {
×
995
          break;
×
996
        }
997
        pOperator->pDownstreamNotifyParams[i] = NULL;
×
998
      }
999
    }
1000
  }
1001
  if (TSDB_CODE_SUCCESS != code) {
×
1002
    pOperator->pTaskInfo->code = code;
×
1003
    T_LONG_JMP(pOperator->pTaskInfo->env, pOperator->pTaskInfo->code);
×
1004
  }
1005

1006
  return code;
×
1007
}
1008

1009
int16_t getOperatorResultBlockId(struct SOperatorInfo* pOperator, int32_t idx) {
222,382✔
1010
  if (pOperator->transparent) {
222,382✔
1011
    return getOperatorResultBlockId(pOperator->pDownstream[idx], 0);
22,070✔
1012
  }
1013
  return pOperator->resultDataBlockId;
200,312✔
1014
}
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