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

taosdata / TDengine / #4824

27 Oct 2025 05:39AM UTC coverage: 61.409% (+0.2%) from 61.242%
#4824

push

travis-ci

web-flow
Merge pull request #33376 from taosdata/3.0

merge 3.0

156919 of 324854 branches covered (48.3%)

Branch coverage included in aggregate %.

371 of 493 new or added lines in 25 files covered. (75.25%)

2822 existing lines in 108 files now uncovered.

208404 of 270043 relevant lines covered (77.17%)

229356519.53 hits per line

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

49.61
/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,
2,147,483,647✔
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 = {
2,147,483,647✔
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
      .resetStateFn = NULL,
47
  };
48

49
  return fpSet;
2,147,483,647✔
50
}
51

52
void setOperatorStreamStateFn(SOperatorInfo* pOperator, __optr_state_fn_t relaseFn, __optr_state_fn_t reloadFn) {
316,588,745✔
53
  pOperator->fpSet.releaseStreamStateFn = relaseFn;
316,588,745✔
54
  pOperator->fpSet.reloadStreamStateFn = reloadFn;
316,741,889✔
55
}
316,609,339✔
56

57
void setOperatorResetStateFn(SOperatorInfo* pOperator, __optr_reset_state_fn_t resetFn) {
2,147,483,647✔
58
  pOperator->fpSet.resetStateFn = resetFn;
2,147,483,647✔
59
}
2,147,483,647✔
60

61
int32_t optrDummyOpenFn(SOperatorInfo* pOperator) {
277,651,954✔
62
  OPTR_SET_OPENED(pOperator);
277,651,954✔
63
  pOperator->cost.openCost = 0;
277,660,830✔
64
  return TSDB_CODE_SUCCESS;
277,678,712✔
65
}
66

67
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num) {
1,055,934,556✔
68
  p->pDownstream = taosMemoryCalloc(1, num * POINTER_BYTES);
1,055,934,556✔
69
  if (p->pDownstream == NULL) {
1,056,256,309!
70
    return terrno;
×
71
  }
72

73
  memcpy(p->pDownstream, pDownstream, num * POINTER_BYTES);
1,055,923,069!
74
  p->numOfDownstream = num;
1,055,942,474✔
75
  p->numOfRealDownstream = num;
1,056,341,492✔
76
  return TSDB_CODE_SUCCESS;
1,055,983,398✔
77
}
78

79
void setOperatorCompleted(SOperatorInfo* pOperator) {
2,147,483,647✔
80
  pOperator->status = OP_EXEC_DONE;
2,147,483,647✔
81
  pOperator->cost.totalCost = (taosGetTimestampUs() - pOperator->pTaskInfo->cost.start) / 1000.0;
2,147,483,647✔
82
  setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
2,147,483,647✔
83
}
2,147,483,647✔
84

85
void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
2,147,483,647✔
86
                     void* pInfo, SExecTaskInfo* pTaskInfo) {
87
  pOperator->name = (char*)name;
2,147,483,647✔
88
  pOperator->operatorType = type;
2,147,483,647✔
89
  pOperator->blocking = blocking;
2,147,483,647✔
90
  pOperator->status = status;
2,147,483,647✔
91
  pOperator->info = pInfo;
2,147,483,647✔
92
  pOperator->pTaskInfo = pTaskInfo;
2,147,483,647✔
93
}
2,147,483,647✔
94

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

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

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

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

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

137
  int64_t t = getQuerySupportBufSize(numOfTables);
×
138

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

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

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

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

157
void traverseOperatorTree(SOperatorInfo* pOperator, optr_fn_t fn, STraverParam* pParam, const char* id) {
64,813,103✔
158
  if (pOperator == NULL) {
64,813,103!
159
    return;
×
160
  }
161

162
  ERetType ret = fn(pOperator, pParam, id);
64,813,103✔
163
  if (ret == OPTR_FN_RET_ABORT || pParam->code != TSDB_CODE_SUCCESS) {
64,812,709✔
164
    return;
34,100,854✔
165
  }
166

167
  for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
61,367,524✔
168
    traverseOperatorTree(pOperator->pDownstream[i], fn, pParam, id);
30,652,036✔
169
    if (pParam->code != 0) {
30,648,450!
170
      break;
×
171
    }
172
  }
173
}
174

175
ERetType extractOperatorInfo(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
64,519,568✔
176
  STraverParam* p = pParam;
64,519,568✔
177
  if (pOperator->operatorType == *(int32_t*)p->pParam) {
64,519,568✔
178
    p->pRet = pOperator;
33,983,917✔
179
    return OPTR_FN_RET_ABORT;
33,986,629✔
180
  } else {
181
    return OPTR_FN_RET_CONTINUE;
30,538,031✔
182
  }
183
}
184

185
// QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN
186
int32_t extractOperatorInTree(SOperatorInfo* pOperator, int32_t type, const char* id, SOperatorInfo** pOptrInfo) {
33,984,079✔
187
  QRY_PARAM_CHECK(pOptrInfo);
33,984,079!
188

189
  if (pOperator == NULL) {
33,989,376!
190
    qError("invalid operator, failed to find tableScanOperator %s", id);
×
191
    return TSDB_CODE_PAR_INTERNAL_ERROR;
×
192
  }
193

194
  STraverParam p = {.pParam = &type, .pRet = NULL};
33,989,376✔
195
  traverseOperatorTree(pOperator, extractOperatorInfo, &p, id);
33,986,952✔
196
  if (p.code == 0) {
33,986,989✔
197
    *pOptrInfo = p.pRet;
33,986,832✔
198
  }
199
  return p.code;
33,985,491✔
200
}
201

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

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

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

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

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

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

256
static ERetType doStopDataReader(SOperatorInfo* pOperator, STraverParam* pParam, const char* pIdStr) {
292,934✔
257
  SStorageAPI* pAPI = pParam->pParam;
292,934✔
258
  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
292,934✔
259
    STableScanInfo* pInfo = pOperator->info;
116,474✔
260

261
    if (pInfo->base.dataReader != NULL) {
116,474!
262
      pAPI->tsdReader.tsdReaderNotifyClosing(pInfo->base.dataReader);
116,474✔
263
    }
264
    return OPTR_FN_RET_ABORT;
116,474✔
265
  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
176,460!
266
    SStreamScanInfo* pInfo = pOperator->info;
×
267

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

275
    return OPTR_FN_RET_ABORT;
×
276
  }
277

278
  return OPTR_FN_RET_CONTINUE;
176,460✔
279
}
280

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

287
int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SNode* pTagCond,
2,147,483,647✔
288
                       SNode* pTagIndexCond, const char* pUser, const char* dbname, SOperatorInfo** pOptrInfo,
289
                       EOPTR_EXEC_MODEL model) {
290
  QRY_PARAM_CHECK(pOptrInfo);
2,147,483,647!
291

292
  int32_t     code = 0;
2,147,483,647✔
293
  int32_t     type = nodeType(pPhyNode);
2,147,483,647✔
294
  const char* idstr = GET_TASKID(pTaskInfo);
2,147,483,647✔
295

296
  if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
2,147,483,647✔
297
    SOperatorInfo* pOperator = NULL;
1,209,856,098✔
298
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
1,209,453,664✔
299
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
598,767,098✔
300
      // NOTE: this is an patch to fix the physical plan
301
      // TODO remove it later
302
      if (pTableScanNode->scan.node.pLimit != NULL) {
598,767,098✔
303
        pTableScanNode->groupSort = true;
21,054,631✔
304
      }
305

306
      STableListInfo* pTableListInfo = tableListCreate();
598,801,322✔
307
      if (!pTableListInfo) {
598,883,256!
308
        pTaskInfo->code = terrno;
×
309
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
310
        return terrno;
×
311
      }
312

313
      // Since virtual stable scan use virtual super table's uid in scan operator, the origin table might be stored on
314
      // different vnode, so we should not get table schema for virtual stable scan.
315
      if (!pTableScanNode->scan.virtualStableScan) {
598,883,256✔
316
        code = initQueriedTableSchemaInfo(pHandle, &pTableScanNode->scan, dbname, pTaskInfo);
586,544,548✔
317
        if (code) {
585,728,209✔
318
          pTaskInfo->code = code;
2,340✔
319
          tableListDestroy(pTableListInfo);
2,340✔
320
          return code;
2,340✔
321
        }
322
      }
323

324
      if (pTableScanNode->scan.node.dynamicOp) {
598,024,613!
325
        pTaskInfo->dynamicTask = true;
37,157,260✔
326
        pTableListInfo->idInfo.suid = pTableScanNode->scan.suid;
37,167,142✔
327
        pTableListInfo->idInfo.tableType = pTableScanNode->scan.tableType;
37,154,053✔
328
      } else {
329
        code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
560,841,383!
330
                                       pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo, NULL);
331
        if (code) {
561,692,587✔
332
          pTaskInfo->code = code;
7,058✔
333
          tableListDestroy(pTableListInfo);
7,058✔
334
          qError("failed to createScanTableListInfo, code:%s, %s", tstrerror(code), idstr);
7,058!
335
          return code;
7,058✔
336
        }
337
      }
338

339
      code = createTableScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
598,850,532✔
340
      if (NULL == pOperator || code != 0) {
598,335,470!
341
        pTaskInfo->code = code;
7,735✔
342
        tableListDestroy(pTableListInfo);
18,558✔
343
        return code;
18,558✔
344
      }
345

346
      STableScanInfo* pScanInfo = pOperator->info;
598,389,195✔
347
      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
598,474,800✔
348
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
610,686,566✔
349
      STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
107,040,293✔
350
      STableListInfo*           pTableListInfo = tableListCreate();
107,040,293✔
351
      if (!pTableListInfo) {
107,196,415!
352
        pTaskInfo->code = terrno;
×
353
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
354
        return terrno;
×
355
      }
356

357
      code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle, pTableListInfo,
107,196,415✔
358
                                     pTagCond, pTagIndexCond, pTaskInfo, NULL);
359
      if (code) {
107,152,220!
360
        pTaskInfo->code = code;
×
361
        tableListDestroy(pTableListInfo);
×
362
        qError("failed to createScanTableListInfo, code:%s", tstrerror(code));
×
363
        return code;
×
364
      }
365

366
      code = initQueriedTableSchemaInfo(pHandle, &pTableScanNode->scan, dbname, pTaskInfo);
107,152,220✔
367
      if (code) {
106,783,098!
368
        pTaskInfo->code = code;
×
369
        tableListDestroy(pTableListInfo);
×
370
        return code;
×
371
      }
372

373
      code = createTableMergeScanOperatorInfo(pTableScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
106,783,098✔
374
      if (NULL == pOperator || code != 0) {
106,935,441!
375
        pTaskInfo->code = code;
60,275✔
376
        tableListDestroy(pTableListInfo);
×
377
        return code;
×
378
      }
379

380
      STableMergeScanInfo* pScanInfo = pOperator->info;
106,875,205✔
381
      pTaskInfo->cost.pRecoder = &pScanInfo->base.readRecorder;
107,021,017✔
382
    } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
503,646,273✔
383
      code = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode,
420,530,783!
384
                                        pTaskInfo, &pOperator);
385
    } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
83,115,490✔
386
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
745,180✔
387
      STableListInfo*      pTableListInfo = tableListCreate();
745,180✔
388
      if (!pTableListInfo) {
745,854!
389
        pTaskInfo->code = terrno;
×
390
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
391
        return terrno;
×
392
      }
393

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

405
      code = createTmqScanOperatorInfo(pHandle, pTableScanNode, pTagCond, pTableListInfo, pTaskInfo, &pOperator);
746,510✔
406
      if (code) {
743,605!
407
        pTaskInfo->code = code;
×
408
        tableListDestroy(pTableListInfo);
×
409
        return code;
×
410
      }
411
    } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
82,370,310✔
412
      SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode;
48,453,577✔
413
      if (pSysScanPhyNode->scan.virtualStableScan) {
48,453,577!
414
        STableListInfo*           pTableListInfo = tableListCreate();
15,245,255✔
415
        if (!pTableListInfo) {
15,246,614!
NEW
416
          pTaskInfo->code = terrno;
×
NEW
417
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
NEW
418
          return terrno;
×
419
        }
420

421
        code = createScanTableListInfo((SScanPhysiNode*)pSysScanPhyNode, NULL, false, pHandle, pTableListInfo, pTagCond,
15,246,614✔
422
                                       pTagIndexCond, pTaskInfo, NULL);
423
        if (code != TSDB_CODE_SUCCESS) {
15,245,241!
NEW
424
          pTaskInfo->code = code;
×
NEW
425
          tableListDestroy(pTableListInfo);
×
NEW
426
          return code;
×
427
        }
428

429
        code = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pTableListInfo, pUser, pTaskInfo, &pOperator);
15,245,241✔
430
      } else {
431
        code = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, NULL, pUser, pTaskInfo, &pOperator);
33,210,900✔
432
      }
433
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN == type) {
33,916,733!
434
      STableCountScanPhysiNode* pTblCountScanNode = (STableCountScanPhysiNode*)pPhyNode;
×
435
      code = createTableCountScanOperatorInfo(pHandle, pTblCountScanNode, pTaskInfo, &pOperator);
×
436
    } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) {
33,916,733✔
437
      STagScanPhysiNode* pTagScanPhyNode = (STagScanPhysiNode*)pPhyNode;
30,114,239✔
438
      STableListInfo*    pTableListInfo = tableListCreate();
30,114,239✔
439
      if (!pTableListInfo) {
30,136,594!
440
        pTaskInfo->code = terrno;
×
441
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
442
        return terrno;
×
443
      }
444

445
      code = initQueriedTableSchemaInfo(pHandle, &pTagScanPhyNode->scan, dbname, pTaskInfo);
30,136,594✔
446
      if (code != TSDB_CODE_SUCCESS) {
30,101,243!
447
        pTaskInfo->code = code;
×
448
        tableListDestroy(pTableListInfo);
×
449
        return code;
×
450
      }
451

452
      if (!pTagScanPhyNode->onlyMetaCtbIdx) {
30,101,243!
453
        code = createScanTableListInfo((SScanPhysiNode*)pTagScanPhyNode, NULL, false, pHandle, pTableListInfo, pTagCond,
5,386,601✔
454
                                       pTagIndexCond, pTaskInfo, NULL);
455
        if (code != TSDB_CODE_SUCCESS) {
5,386,601!
456
          pTaskInfo->code = code;
×
457
          qError("failed to getTableList, code:%s", tstrerror(code));
×
458
          tableListDestroy(pTableListInfo);
×
459
          return code;
×
460
        }
461
      }
462

463
      if (pTagScanPhyNode->scan.node.dynamicOp) {
30,101,268!
464
        pTaskInfo->dynamicTask = true;
4,061,108✔
465
        pTableListInfo->idInfo.suid = pTagScanPhyNode->scan.suid;
4,061,108✔
466
        pTableListInfo->idInfo.tableType = pTagScanPhyNode->scan.tableType;
4,061,108✔
467
      }
468

469
      code = createTagScanOperatorInfo(pHandle, pTagScanPhyNode, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo,
30,103,548✔
470
                                       &pOperator);
471
      if (code) {
30,112,532!
472
        pTaskInfo->code = code;
×
473
        tableListDestroy(pTableListInfo);
×
474
        return code;
×
475
      }
476
    } else if (QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN == type) {
3,802,494✔
477
      SBlockDistScanPhysiNode* pBlockNode = (SBlockDistScanPhysiNode*)pPhyNode;
21,539✔
478
      STableListInfo*          pTableListInfo = tableListCreate();
21,539✔
479
      if (!pTableListInfo) {
21,539!
480
        pTaskInfo->code = terrno;
×
481
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
482
        return terrno;
×
483
      }
484

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

495
        size_t num = taosArrayGetSize(pList);
17,955✔
496
        for (int32_t i = 0; i < num; ++i) {
42,521✔
497
          uint64_t* id = taosArrayGet(pList, i);
24,566✔
498
          if (id == NULL) {
24,566!
499
            continue;
×
500
          }
501

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

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

521
      code = createDataBlockInfoScanOperator(pHandle, pBlockNode, pTableListInfo, pTaskInfo, &pOperator);
21,539✔
522
      if (code) {
21,539!
523
        pTaskInfo->code = code;
×
524
        tableListDestroy(pTableListInfo);
×
525
        return code;
×
526
      }
527
    } else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) {
3,780,955✔
528
      SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode;
2,910,464✔
529
      STableListInfo*        pTableListInfo = tableListCreate();
2,910,464✔
530
      if (!pTableListInfo) {
2,911,056!
531
        pTaskInfo->code = terrno;
×
532
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
533
        return terrno;
×
534
      }
535

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

544
      code = initQueriedTableSchemaInfo(pHandle, &pScanNode->scan, dbname, pTaskInfo);
2,911,453✔
545
      if (code != TSDB_CODE_SUCCESS) {
2,908,200!
546
        pTaskInfo->code = code;
×
547
        tableListDestroy(pTableListInfo);
×
548
        return code;
×
549
      }
550

551
      code = createCacherowsScanOperator(pScanNode, pHandle, pTableListInfo, pTaskInfo, &pOperator);
2,908,200✔
552
      if (code) {
2,910,128!
553
        tableListDestroy(pTableListInfo);
×
554
        pTaskInfo->code = code;
×
555
        return code;
×
556
      }
557
    } else if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
870,491✔
558
      code = createProjectOperatorInfo(NULL, (SProjectPhysiNode*)pPhyNode, pTaskInfo, &pOperator);
842,284✔
559
    } else if (QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN == type) {
28,207!
560
      // NOTE: this is an patch to fix the physical plan
561
      code = createVirtualTableMergeOperatorInfo(NULL, 0, (SVirtualScanPhysiNode*)pPhyNode, pTaskInfo, &pOperator);
42,382✔
562
    } else {
563
      code = TSDB_CODE_INVALID_PARA;
×
564
      pTaskInfo->code = code;
×
565
      return code;
×
566
    }
567

568
    if (pOperator != NULL) {  // todo moved away
1,208,935,074✔
569
      pOperator->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
1,208,939,294✔
570
    }
571

572
    *pOptrInfo = pOperator;
1,209,435,061✔
573
    return code;
1,209,180,399✔
574
  }
575

576
  size_t          size = LIST_LENGTH(pPhyNode->pChildren);
1,056,415,711!
577
  SOperatorInfo** ops = taosMemoryCalloc(size, POINTER_BYTES);
1,056,583,194✔
578
  if (ops == NULL) {
1,056,497,649!
579
    code = terrno;
×
580
    pTaskInfo->code = code;
×
581
    return code;
×
582
  }
583

584
  for (int32_t i = 0; i < size; ++i) {
2,147,483,647✔
585
    SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, i);
1,210,839,606✔
586
    code = createOperator(pChildNode, pTaskInfo, pHandle, pTagCond, pTagIndexCond, pUser, dbname, &ops[i], model);
1,210,863,129✔
587
    if (ops[i] == NULL || code != 0) {
1,210,508,266✔
588
      for (int32_t j = 0; j < i; ++j) {
19,452!
589
        destroyOperator(ops[j]);
×
590
      }
591
      taosMemoryFree(ops);
19,452!
592
      return code;
19,452✔
593
    }
594
  }
595

596
  SOperatorInfo* pOptr = NULL;
1,055,861,319✔
597
  if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
1,055,874,053✔
598
    code = createProjectOperatorInfo(ops[0], (SProjectPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
315,565,240✔
599
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_AGG == type) {
740,308,813✔
600
    SAggPhysiNode* pAggNode = (SAggPhysiNode*)pPhyNode;
412,734,358✔
601
    if (pAggNode->pGroupKeys != NULL) {
412,734,358✔
602
      code = createGroupOperatorInfo(ops[0], pAggNode, pTaskInfo, &pOptr);
109,919,421✔
603
    } else {
604
      code = createAggregateOperatorInfo(ops[0], pAggNode, pTaskInfo, &pOptr);
302,959,101✔
605
    }
606
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type) {
327,574,455✔
607
    SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
15,872,029✔
608
    code = createIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
15,872,029✔
609
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) {
311,702,426✔
610
    SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode;
3,193,571✔
611
    code = createMergeAlignedIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
3,193,571✔
612
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL == type) {
308,508,855!
613
    SMergeIntervalPhysiNode* pIntervalPhyNode = (SMergeIntervalPhysiNode*)pPhyNode;
×
614
    code = createMergeIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, &pOptr);
×
615
  } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
308,508,855✔
616
    code = createSortOperatorInfo(ops[0], (SSortPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
134,013,096✔
617
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT == type) {
174,495,759✔
618
    code = createGroupSortOperatorInfo(ops[0], (SGroupSortPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
113,420✔
619
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) {
174,382,339✔
620
    SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode;
52,166,428✔
621
    code = createMultiwayMergeOperatorInfo(ops, size, pMergePhyNode, pTaskInfo, &pOptr);
52,166,428✔
622
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION == type) {
122,215,911✔
623
    SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
1,195,795✔
624
    code = createSessionAggOperatorInfo(ops[0], pSessionNode, pTaskInfo, &pOptr);
1,195,795✔
625
  } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
121,020,116✔
626
    code = createPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
9,612,155✔
627
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE == type) {
111,407,961✔
628
    SStateWindowPhysiNode* pStateNode = (SStateWindowPhysiNode*)pPhyNode;
1,508,752✔
629
    code = createStatewindowOperatorInfo(ops[0], pStateNode, pTaskInfo, &pOptr);
1,508,752✔
630
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN == type) {
109,899,209✔
631
    code = createMergeJoinOperatorInfo(ops, size, (SSortMergeJoinPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
52,678,743✔
632
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_JOIN == type) {
57,220,466✔
633
    code = createHashJoinOperatorInfo(ops, size, (SHashJoinPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
6,169,816✔
634
  } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) {
51,050,650✔
635
    code = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
2,584,697✔
636
  } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) {
48,465,953✔
637
    code = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
12,635,353✔
638
  } else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) {
35,830,616✔
639
    code = createTimeSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
11,917,805✔
640
  } else if (QUERY_NODE_PHYSICAL_PLAN_FORECAST_FUNC == type) {
23,912,811!
641
    code = createForecastOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
642
  } else if (QUERY_NODE_PHYSICAL_PLAN_IMPUTATION_FUNC == type) {
23,912,811!
643
    code = createImputationOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
644
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT == type) {
23,912,811✔
645
    code = createEventwindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
444,887✔
646
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_CACHE == type) {
23,467,924✔
647
    code = createGroupCacheOperatorInfo(ops, size, (SGroupCachePhysiNode*)pPhyNode, pTaskInfo, &pOptr);
6,169,816✔
648
  } else if (QUERY_NODE_PHYSICAL_PLAN_DYN_QUERY_CTRL == type) {
17,298,108✔
649
    code = createDynQueryCtrlOperatorInfo(ops, size, (SDynQueryCtrlPhysiNode*)pPhyNode, pTaskInfo, pHandle->pMsgCb, &pOptr);
8,718,123✔
650
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_COUNT == type) {
8,579,985✔
651
    code = createCountwindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
378,390✔
652
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ANOMALY == type) {
8,201,595!
653
    code = createAnomalywindowOperatorInfo(ops[0], pPhyNode, pTaskInfo, &pOptr);
×
654
  } else if (QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN == type) {
8,201,595✔
655
    SVirtualScanPhysiNode* pVirtualTableScanNode = (SVirtualScanPhysiNode*)pPhyNode;
7,976,252✔
656
    // NOTE: this is an patch to fix the physical plan
657

658
    if (pVirtualTableScanNode->scan.node.pLimit != NULL) {
7,976,252!
659
      pVirtualTableScanNode->groupSort = true;
×
660
    }
661
    code = createVirtualTableMergeOperatorInfo(ops, size, (SVirtualScanPhysiNode*)pPhyNode, pTaskInfo, &pOptr);
7,976,252✔
662
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_EXTERNAL == type) {
225,621✔
663
    code = createExternalWindowOperator(ops[0], pPhyNode, pTaskInfo, &pOptr);
236,775✔
664
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_EXTERNAL == type) {
176!
665
    code = createMergeAlignedExternalWindowOperator(ops[0], pPhyNode, pTaskInfo, &pOptr);
4,636✔
666
  } else {
667
    code = TSDB_CODE_INVALID_PARA;
×
668
    pTaskInfo->code = code;
×
669
    for (int32_t i = 0; i < size; ++i) {
×
670
      destroyOperator(ops[i]);
×
671
    }
672
    taosMemoryFree(ops);
×
673
    qError("invalid operator type %d", type);
×
674
    return code;
×
675
  }
676

677
  taosMemoryFree(ops);
1,055,976,232✔
678
  if (pOptr) {
1,055,870,775✔
679
    pOptr->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
1,055,708,891✔
680
  }
681

682
  *pOptrInfo = pOptr;
1,056,087,647✔
683
  return code;
1,056,041,028✔
684
}
685

686
void destroyOperator(SOperatorInfo* pOperator) {
2,147,483,647✔
687
  if (pOperator == NULL) {
2,147,483,647✔
688
    return;
27,956✔
689
  }
690

691
  freeResetOperatorParams(pOperator, OP_GET_PARAM, true);
2,147,483,647✔
692
  freeResetOperatorParams(pOperator, OP_NOTIFY_PARAM, true);
2,147,483,647✔
693

694
  if (pOperator->pDownstream != NULL) {
2,147,483,647✔
695
    for (int32_t i = 0; i < pOperator->numOfRealDownstream; ++i) {
2,147,483,647✔
696
      destroyOperator(pOperator->pDownstream[i]);
1,211,134,139✔
697
    }
698

699
    taosMemoryFreeClear(pOperator->pDownstream);
1,056,817,793!
700
    pOperator->numOfDownstream = 0;
1,056,705,018✔
701
  }
702

703
  if (pOperator->fpSet.closeFn != NULL && pOperator->info != NULL) {
2,147,483,647✔
704
    pOperator->fpSet.closeFn(pOperator->info);
2,147,483,647✔
705
  }
706

707
  cleanupExprSupp(&pOperator->exprSupp);
2,147,483,647✔
708
  taosMemoryFreeClear(pOperator);
2,147,483,647!
709
}
710

711
void destroyOperatorAndDownstreams(SOperatorInfo* pOperator, SOperatorInfo** downstreams, int32_t num) {
×
712
  if (downstreams != NULL) {
×
713
    for (int i = 0; i < num; i++) {
×
714
      destroyOperator(downstreams[i]);
×
715
    }
716
  }
717

718
  if (pOperator != NULL) {
×
719
    pOperator->info = NULL;
×
720
    if (pOperator->pDownstream != NULL) {
×
721
      taosMemoryFreeClear(pOperator->pDownstream);
×
722
      pOperator->pDownstream = NULL;
×
723
    }
724
    destroyOperator(pOperator);
×
725
  }
726
}
×
727

728
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
33,090,823✔
729
  SExplainExecInfo  execInfo = {0};
33,090,823✔
730
  SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
33,090,823✔
731
  if (pExplainInfo == NULL) {
33,090,823!
732
    return terrno;
×
733
  }
734

735
  pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
33,090,823✔
736
  pExplainInfo->startupCost = operatorInfo->cost.openCost;
33,090,823✔
737
  pExplainInfo->totalCost = operatorInfo->cost.totalCost;
33,090,589✔
738
  pExplainInfo->verboseLen = 0;
33,090,589✔
739
  pExplainInfo->verboseInfo = NULL;
33,090,589✔
740

741
  if (operatorInfo->fpSet.getExplainFn) {
33,090,589✔
742
    int32_t code =
743
        operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
13,466,931✔
744
    if (code) {
13,465,720!
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;
33,089,378✔
751
  for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
51,604,200✔
752
    code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
18,515,056✔
753
    if (code != TSDB_CODE_SUCCESS) {
18,514,822!
754
      //      taosMemoryFreeClear(*pRes);
755
      return code;
×
756
    }
757
  }
758

759
  return TSDB_CODE_SUCCESS;
33,090,589✔
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) {
201,836,118✔
836
  SOperatorParam**  ppParam = NULL;
201,836,118✔
837
  SOperatorParam*** pppDownstramParam = NULL;
201,836,118✔
838
  switch (type) {
201,836,118!
839
    case OP_GET_PARAM:
201,836,118✔
840
      ppParam = &pOperator->pOperatorGetParam;
201,836,118✔
841
      pppDownstramParam = &pOperator->pDownstreamGetParams;
201,836,118✔
842
      break;
201,836,118✔
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);
201,836,118✔
852

853
  if (NULL == pInput) {
201,834,964!
854
    return TSDB_CODE_SUCCESS;
×
855
  }
856

857
  *ppParam = (pInput->opType == pOperator->operatorType) ? pInput : NULL;
201,834,964!
858

859
  if (NULL == *pppDownstramParam) {
201,834,964✔
860
    *pppDownstramParam = taosMemoryCalloc(pOperator->numOfDownstream, POINTER_BYTES);
19,965,715!
861
    if (NULL == *pppDownstramParam) {
19,965,715!
862
      return terrno;
×
863
    }
864
  }
865

866
  if (NULL == *ppParam) {
201,836,118!
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);
201,834,964!
874

875
  int32_t childrenNum = taosArrayGetSize((*ppParam)->pChildren);
201,836,118✔
876
  if (childrenNum <= 0) {
201,834,964✔
877
    return TSDB_CODE_SUCCESS;
185,571,238✔
878
  }
879

880
  for (int32_t i = 0; i < childrenNum; ++i) {
48,791,178✔
881
    SOperatorParam* pChild = *(SOperatorParam**)taosArrayGet((*ppParam)->pChildren, i);
32,527,452✔
882
    if (pChild == NULL) {
32,527,452!
883
      return terrno;
×
884
    }
885

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

896
  taosArrayDestroy((*ppParam)->pChildren);
16,263,726✔
897
  (*ppParam)->pChildren = NULL;
16,263,726✔
898

899
  return TSDB_CODE_SUCCESS;
16,263,726✔
900
}
901

902
SSDataBlock* getNextBlockFromDownstream(struct SOperatorInfo* pOperator, int32_t idx) {
2,147,483,647✔
903
  SSDataBlock* p = NULL;
2,147,483,647✔
904
  int32_t      code = getNextBlockFromDownstreamImpl(pOperator, idx, true, &p);
2,147,483,647✔
905
  if (code == TSDB_CODE_SUCCESS) {
2,147,483,647!
906
    code = blockDataCheck(p);
2,147,483,647✔
907
    if (code != TSDB_CODE_SUCCESS) {
2,147,483,647!
908
      qError("blockDataCheck failed, code:%s", tstrerror(code));
×
909
    }
910
  }
911
  return (code == 0) ? p : NULL;
2,147,483,647!
912
}
913

914
SSDataBlock* getNextBlockFromDownstreamRemain(struct SOperatorInfo* pOperator, int32_t idx) {
275,792,085✔
915
  SSDataBlock* p = NULL;
275,792,085✔
916
  int32_t      code = getNextBlockFromDownstreamImpl(pOperator, idx, false, &p);
275,793,397✔
917
  if (code == TSDB_CODE_SUCCESS) {
275,783,877!
918
    code = blockDataCheck(p);
275,786,827✔
919
    if (code != TSDB_CODE_SUCCESS) {
275,780,686!
920
      qError("blockDataCheck failed, code:%s", tstrerror(code));
×
921
    }
922
  }
923
  return (code == 0)? p:NULL;
275,781,541!
924
}
925

926
int32_t optrDefaultGetNextExtFn(struct SOperatorInfo* pOperator, SOperatorParam* pParam, SSDataBlock** pRes) {
201,836,118✔
927
  QRY_PARAM_CHECK(pRes);
201,836,118!
928

929
  int32_t lino = 0;
201,836,118✔
930
  int32_t code = setOperatorParams(pOperator, pParam, OP_GET_PARAM);
201,836,118✔
931
  QUERY_CHECK_CODE(code, lino, _end);
201,834,964!
932
  code = pOperator->fpSet.getNextFn(pOperator, pRes);
201,834,964✔
933
  QUERY_CHECK_CODE(code, lino, _end);
201,834,907!
934

935
_end:
201,834,907✔
936
  if (code != TSDB_CODE_SUCCESS) {
201,834,907!
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;
201,834,907✔
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) {
117,971,691✔
970
  if (pOperator->transparent) {
117,971,691!
971
    return getOperatorResultBlockId(pOperator->pDownstream[idx], 0);
12,339,632✔
972
  }
973
  return pOperator->resultDataBlockId;
105,633,783✔
974
}
975

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

980
void resetBasicOperatorState(SOptrBasicInfo *pBasicInfo) {
78,334,473✔
981
  if (pBasicInfo->pRes) blockDataCleanup(pBasicInfo->pRes);
78,334,473!
982
  initResultRowInfo(&pBasicInfo->resultRowInfo);
78,328,679✔
983
}
78,332,149✔
984

985
int32_t resetAggSup(SExprSupp* pExprSupp, SAggSupporter* pSup, SExecTaskInfo* pTaskInfo,
64,267,851✔
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;
64,267,851✔
989
  SExprInfo* pExprInfo = NULL;
64,267,851✔
990
  cleanupAggSup(pSup);
64,267,851✔
991
  cleanupExprSuppWithoutFilter(pExprSupp);
64,264,394✔
992
  code = createExprInfo(pNodeList, pGroupKeys, &pExprInfo, &num);
64,258,528✔
993
  QUERY_CHECK_CODE(code, lino, _error);
64,267,851!
994
  code = initAggSup(pExprSupp, pSup, pExprInfo, num, keyBufSize, pKey, pState, pStore);
64,267,851✔
995
  QUERY_CHECK_CODE(code, lino, _error);
64,265,564!
996
  return code;
64,265,564✔
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,
24,221,633✔
1006
                      SNodeList* pGroupKeys, SFunctionStateStore* pStore) {
1007
  int32_t code = 0, lino = 0, num = 0;
24,221,633✔
1008
  SExprInfo* pExprInfo = NULL;
24,221,633✔
1009
  cleanupExprSuppWithoutFilter(pExprSupp);
24,221,633✔
1010
  code = createExprInfo(pNodeList, pGroupKeys, &pExprInfo, &num);
24,219,309✔
1011
  QUERY_CHECK_CODE(code, lino, _error);
24,219,309!
1012
  code = initExprSupp(pExprSupp, pExprInfo, num, pStore);
24,219,309✔
1013
  QUERY_CHECK_CODE(code, lino, _error);
24,219,309!
1014
  return code;
24,219,309✔
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
}
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