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

taosdata / TDengine / #4821

22 Oct 2025 06:02AM UTC coverage: 61.242% (-0.1%) from 61.353%
#4821

push

travis-ci

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

fix(stream): reset tableScan operator (#33225)

156089 of 324573 branches covered (48.09%)

Branch coverage included in aggregate %.

184 of 244 new or added lines in 9 files covered. (75.41%)

789 existing lines in 124 files now uncovered.

207891 of 269759 relevant lines covered (77.07%)

244003752.29 hits per line

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

59.95
/source/libs/executor/src/executil.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 "function.h"
17
#include "functionMgt.h"
18
#include "index.h"
19
#include "os.h"
20
#include "query.h"
21
#include "querynodes.h"
22
#include "tarray.h"
23
#include "tdatablock.h"
24
#include "thash.h"
25
#include "tmsg.h"
26
#include "ttime.h"
27

28
#include "executil.h"
29
#include "executorInt.h"
30
#include "querytask.h"
31
#include "storageapi.h"
32
#include "tcompression.h"
33

34
typedef struct tagFilterAssist {
35
  SHashObj* colHash;
36
  int32_t   index;
37
  SArray*   cInfoList;
38
  int32_t   code;
39
} tagFilterAssist;
40

41
typedef struct STransTagExprCtx {
42
  int32_t      code;
43
  SMetaReader* pReader;
44
} STransTagExprCtx;
45

46
typedef enum {
47
  FILTER_NO_LOGIC = 1,
48
  FILTER_AND,
49
  FILTER_OTHER,
50
} FilterCondType;
51

52
static FilterCondType checkTagCond(SNode* cond);
53
static int32_t optimizeTbnameInCond(void* metaHandle, int64_t suid, SArray* list, SNode* pTagCond, SStorageAPI* pAPI);
54
static int32_t optimizeTbnameInCondImpl(void* metaHandle, SArray* list, SNode* pTagCond, SStorageAPI* pStoreAPI,
55
                                        uint64_t suid);
56

57
static int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
58
                            STableListInfo* pListInfo, uint8_t* digest, const char* idstr, SStorageAPI* pStorageAPI, void* pStreamInfo);
59

60
static int64_t getLimit(const SNode* pLimit) {
2,147,483,647✔
61
  return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->limit) ? -1 : ((SLimitNode*)pLimit)->limit->datum.i;
2,147,483,647✔
62
}
63
static int64_t getOffset(const SNode* pLimit) {
2,147,483,647✔
64
  return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->offset) ? -1 : ((SLimitNode*)pLimit)->offset->datum.i;
2,147,483,647✔
65
}
66
static void releaseColInfoData(void* pCol);
67

68
void initResultRowInfo(SResultRowInfo* pResultRowInfo) {
1,657,654,001✔
69
  pResultRowInfo->size = 0;
1,657,654,001✔
70
  pResultRowInfo->cur.pageId = -1;
1,657,747,655✔
71
}
1,657,817,055✔
72

73
void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }
15,308,881✔
74

75
void resetResultRow(SResultRow* pResultRow, size_t entrySize) {
2,147,483,647✔
76
  pResultRow->numOfRows = 0;
2,147,483,647✔
77
  pResultRow->closed = false;
2,147,483,647✔
78
  pResultRow->endInterp = false;
2,147,483,647✔
79
  pResultRow->startInterp = false;
2,147,483,647✔
80

81
  if (entrySize > 0) {
2,147,483,647!
82
    memset(pResultRow->pEntryInfo, 0, entrySize);
2,147,483,647!
83
  }
84
}
2,147,483,647✔
85

86
// TODO refactor: use macro
87
SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) {
2,147,483,647✔
88
  return (SResultRowEntryInfo*)((char*)pRow->pEntryInfo + offset[index]);
2,147,483,647✔
89
}
90

91
size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
1,013,756,425✔
92
  int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow);
1,013,756,425✔
93

94
  for (int32_t i = 0; i < numOfOutput; ++i) {
2,147,483,647✔
95
    rowSize += pCtx[i].resDataInfo.interBufSize;
2,147,483,647✔
96
  }
97

98
  return rowSize;
1,013,680,164✔
99
}
100

101
// Convert buf read from rocksdb to result row
102
int32_t getResultRowFromBuf(SExprSupp* pSup, const char* inBuf, size_t inBufSize, char** outBuf, size_t* outBufSize) {
×
103
  if (inBuf == NULL || pSup == NULL) {
×
104
    qError("invalid input parameters, inBuf:%p, pSup:%p", inBuf, pSup);
×
105
    return TSDB_CODE_INVALID_PARA;
×
106
  }
107
  SqlFunctionCtx* pCtx = pSup->pCtx;
×
108
  int32_t*        offset = pSup->rowEntryInfoOffset;
×
109
  SResultRow*     pResultRow = NULL;
×
110
  size_t          processedSize = 0;
×
111
  int32_t         code = TSDB_CODE_SUCCESS;
×
112

113
  // calculate the size of output buffer
114
  *outBufSize = getResultRowSize(pCtx, pSup->numOfExprs);
×
115
  *outBuf = taosMemoryMalloc(*outBufSize);
×
116
  if (*outBuf == NULL) {
×
117
    qError("failed to allocate memory for output buffer, size:%zu", *outBufSize);
×
118
    return terrno;
×
119
  }
120
  pResultRow = (SResultRow*)*outBuf;
×
121
  (void)memcpy(pResultRow, inBuf, sizeof(SResultRow));
×
122
  inBuf += sizeof(SResultRow);
×
123
  processedSize += sizeof(SResultRow);
×
124

125
  for (int32_t i = 0; i < pSup->numOfExprs; ++i) {
×
126
    int32_t len = *(int32_t*)inBuf;
×
127
    inBuf += sizeof(int32_t);
×
128
    processedSize += sizeof(int32_t);
×
129
    if (pResultRow->version != FUNCTION_RESULT_INFO_VERSION && pCtx->fpSet.decode) {
×
130
      code = pCtx->fpSet.decode(&pCtx[i], inBuf, getResultEntryInfo(pResultRow, i, offset), pResultRow->version);
×
131
      if (code != TSDB_CODE_SUCCESS) {
×
132
        qError("failed to decode result row, code:%d", code);
×
133
        return code;
×
134
      }
135
    } else {
136
      (void)memcpy(getResultEntryInfo(pResultRow, i, offset), inBuf, len);
×
137
    }
138
    inBuf += len;
×
139
    processedSize += len;
×
140
  }
141

142
  if (processedSize < inBufSize) {
×
143
    // stream stores extra data after result row
144
    size_t leftLen = inBufSize - processedSize;
×
145
    TAOS_MEMORY_REALLOC(*outBuf, *outBufSize + leftLen);
×
146
    if (*outBuf == NULL) {
×
147
      qError("failed to reallocate memory for output buffer, size:%zu", *outBufSize + leftLen);
×
148
      return terrno;
×
149
    }
150
    (void)memcpy(*outBuf + *outBufSize, inBuf, leftLen);
×
151
    inBuf += leftLen;
×
152
    processedSize += leftLen;
×
153
    *outBufSize += leftLen;
×
154
  }
155

156
  qTrace("[StreamInternal] get result inBufSize:%zu, outBufSize:%zu", inBufSize, *outBufSize);
×
157
  return TSDB_CODE_SUCCESS;
×
158
}
159

160
// Convert result row to buf for rocksdb
161
int32_t putResultRowToBuf(SExprSupp* pSup, const char* inBuf, size_t inBufSize, char** outBuf, size_t* outBufSize) {
×
162
  if (pSup == NULL || inBuf == NULL || outBuf == NULL || outBufSize == NULL) {
×
163
    qError("invalid input parameters, inBuf:%p, pSup:%p, outBufSize:%p, outBuf:%p", inBuf, pSup, outBufSize, outBuf);
×
164
    return TSDB_CODE_INVALID_PARA;
×
165
  }
166

167
  SqlFunctionCtx* pCtx = pSup->pCtx;
×
168
  int32_t*        offset = pSup->rowEntryInfoOffset;
×
169
  SResultRow*     pResultRow = (SResultRow*)inBuf;
×
170
  size_t          rowSize = getResultRowSize(pCtx, pSup->numOfExprs);
×
171

172
  if (rowSize > inBufSize) {
×
173
    qError("invalid input buffer size, rowSize:%zu, inBufSize:%zu", rowSize, inBufSize);
×
174
    return TSDB_CODE_INVALID_PARA;
×
175
  }
176

177
  // calculate the size of output buffer
178
  *outBufSize = rowSize + sizeof(int32_t) * pSup->numOfExprs;
×
179
  if (rowSize < inBufSize) {
×
180
    *outBufSize += inBufSize - rowSize;
×
181
  }
182

183
  *outBuf = taosMemoryMalloc(*outBufSize);
×
184
  if (*outBuf == NULL) {
×
185
    qError("failed to allocate memory for output buffer, size:%zu", *outBufSize);
×
186
    return terrno;
×
187
  }
188

189
  char* pBuf = *outBuf;
×
190
  pResultRow->version = FUNCTION_RESULT_INFO_VERSION;
×
191
  (void)memcpy(pBuf, pResultRow, sizeof(SResultRow));
×
192
  pBuf += sizeof(SResultRow);
×
193
  for (int32_t i = 0; i < pSup->numOfExprs; ++i) {
×
194
    size_t len = sizeof(SResultRowEntryInfo) + pCtx[i].resDataInfo.interBufSize;
×
195
    *(int32_t*)pBuf = (int32_t)len;
×
196
    pBuf += sizeof(int32_t);
×
197
    (void)memcpy(pBuf, getResultEntryInfo(pResultRow, i, offset), len);
×
198
    pBuf += len;
×
199
  }
200

201
  if (rowSize < inBufSize) {
×
202
    // stream stores extra data after result row
203
    size_t leftLen = inBufSize - rowSize;
×
204
    (void)memcpy(pBuf, inBuf + rowSize, leftLen);
×
205
    pBuf += leftLen;
×
206
  }
207

208
  qTrace("[StreamInternal] put result inBufSize:%zu, outBufSize:%zu", inBufSize, *outBufSize);
×
209
  return TSDB_CODE_SUCCESS;
×
210
}
211

212
static void freeEx(void* p) { taosMemoryFree(*(void**)p); }
×
213

214
void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
558,494,350✔
215
  taosMemoryFreeClear(pGroupResInfo->pBuf);
558,494,350✔
216
  if (pGroupResInfo->freeItem) {
558,500,454!
217
    //    taosArrayDestroy(pGroupResInfo->pRows);
218
    taosArrayDestroyEx(pGroupResInfo->pRows, freeEx);
×
219
    pGroupResInfo->freeItem = false;
×
220
    pGroupResInfo->pRows = NULL;
×
221
  } else {
222
    taosArrayDestroy(pGroupResInfo->pRows);
558,500,161✔
223
    pGroupResInfo->pRows = NULL;
558,494,633✔
224
  }
225
  pGroupResInfo->index = 0;
558,501,662✔
226
  pGroupResInfo->delIndex = 0;
558,500,976✔
227
}
558,498,051✔
228

229
int32_t resultrowComparAsc(const void* p1, const void* p2) {
2,147,483,647✔
230
  SResKeyPos* pp1 = *(SResKeyPos**)p1;
2,147,483,647✔
231
  SResKeyPos* pp2 = *(SResKeyPos**)p2;
2,147,483,647✔
232

233
  if (pp1->groupId == pp2->groupId) {
2,147,483,647✔
234
    int64_t pts1 = *(int64_t*)pp1->key;
2,147,483,647✔
235
    int64_t pts2 = *(int64_t*)pp2->key;
2,147,483,647✔
236

237
    if (pts1 == pts2) {
2,147,483,647!
238
      return 0;
×
239
    } else {
240
      return pts1 < pts2 ? -1 : 1;
2,147,483,647✔
241
    }
242
  } else {
243
    return pp1->groupId < pp2->groupId ? -1 : 1;
2,147,483,647✔
244
  }
245
}
246

247
static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); }
2,147,483,647✔
248

249
int32_t initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order) {
444,869,695✔
250
  int32_t code = TSDB_CODE_SUCCESS;
444,869,695✔
251
  int32_t lino = 0;
444,869,695✔
252
  if (pGroupResInfo->pRows != NULL) {
444,869,695✔
253
    taosArrayDestroy(pGroupResInfo->pRows);
51,918,770✔
254
  }
255
  if (pGroupResInfo->pBuf) {
444,880,619✔
256
    taosMemoryFree(pGroupResInfo->pBuf);
51,918,770!
257
    pGroupResInfo->pBuf = NULL;
51,918,770✔
258
  }
259

260
  // extract the result rows information from the hash map
261
  int32_t size = tSimpleHashGetSize(pHashmap);
444,872,509✔
262

263
  void* pData = NULL;
444,865,137✔
264
  pGroupResInfo->pRows = taosArrayInit(size, POINTER_BYTES);
444,865,137✔
265
  QUERY_CHECK_NULL(pGroupResInfo->pRows, code, lino, _end, terrno);
444,874,131!
266

267
  size_t  keyLen = 0;
444,863,762✔
268
  int32_t iter = 0;
444,865,268✔
269
  int64_t bufLen = 0, offset = 0;
444,873,098✔
270

271
  // todo move away and record this during create window
272
  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
2,147,483,647✔
273
    /*void* key = */ (void)tSimpleHashGetKey(pData, &keyLen);
274
    bufLen += keyLen + sizeof(SResultRowPosition);
2,147,483,647✔
275
  }
276

277
  pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);
442,801,120✔
278
  QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
444,876,721!
279

280
  iter = 0;
444,870,769✔
281
  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
2,147,483,647✔
282
    void* key = tSimpleHashGetKey(pData, &keyLen);
2,147,483,647✔
283

284
    SResKeyPos* p = (SResKeyPos*)(pGroupResInfo->pBuf + offset);
2,147,483,647✔
285

286
    p->groupId = *(uint64_t*)key;
2,147,483,647✔
287
    p->pos = *(SResultRowPosition*)pData;
2,147,483,647✔
288
    memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t));
2,147,483,647!
289
    void* tmp = taosArrayPush(pGroupResInfo->pRows, &p);
2,147,483,647✔
290
    QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
2,147,483,647!
291

292
    offset += keyLen + sizeof(struct SResultRowPosition);
2,147,483,647✔
293
  }
294

295
  if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
444,280,059✔
296
    __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc;
23,845,536✔
297
    size = POINTER_BYTES;
23,845,536✔
298
    taosSort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), size, fn);
23,845,536✔
299
  }
300

301
  pGroupResInfo->index = 0;
444,281,472✔
302

303
_end:
444,774,574✔
304
  if (code != TSDB_CODE_SUCCESS) {
444,870,587!
305
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
306
  }
307
  return code;
444,870,587✔
308
}
309

310
void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList) {
×
311
  if (pGroupResInfo->pRows != NULL) {
×
312
    taosArrayDestroy(pGroupResInfo->pRows);
×
313
  }
314

315
  pGroupResInfo->freeItem = true;
×
316
  pGroupResInfo->pRows = pArrayList;
×
317
  pGroupResInfo->index = 0;
×
318
  pGroupResInfo->delIndex = 0;
×
319
}
×
320

321
bool hasRemainResults(SGroupResInfo* pGroupResInfo) {
1,852,966,257✔
322
  if (pGroupResInfo->pRows == NULL) {
1,852,966,257!
323
    return false;
×
324
  }
325

326
  return pGroupResInfo->index < taosArrayGetSize(pGroupResInfo->pRows);
1,852,980,217✔
327
}
328

329
int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {
980,518,206✔
330
  if (pGroupResInfo->pRows == 0) {
980,518,206!
331
    return 0;
×
332
  }
333

334
  return (int32_t)taosArrayGetSize(pGroupResInfo->pRows);
980,529,105✔
335
}
336

337
SArray* createSortInfo(SNodeList* pNodeList) {
239,392,128✔
338
  size_t numOfCols = 0;
239,392,128✔
339

340
  if (pNodeList != NULL) {
239,392,128✔
341
    numOfCols = LIST_LENGTH(pNodeList);
238,908,578✔
342
  } else {
343
    numOfCols = 0;
483,550✔
344
  }
345

346
  SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));
239,388,127✔
347
  if (pList == NULL) {
239,381,100!
348
    return pList;
×
349
  }
350

351
  for (int32_t i = 0; i < numOfCols; ++i) {
530,667,095✔
352
    SOrderByExprNode* pSortKey = (SOrderByExprNode*)nodesListGetNode(pNodeList, i);
291,274,205✔
353
    if (!pSortKey) {
291,280,892!
354
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
355
      taosArrayDestroy(pList);
×
356
      pList = NULL;
×
357
      terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
358
      break;
×
359
    }
360
    SBlockOrderInfo bi = {0};
291,280,892✔
361
    bi.order = (pSortKey->order == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
291,275,786✔
362
    bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST);
291,277,762✔
363

364
    if (nodeType(pSortKey->pExpr) != QUERY_NODE_COLUMN) {
291,279,040!
365
      qError("invalid order by expr type:%d", nodeType(pSortKey->pExpr));
×
366
      taosArrayDestroy(pList);
×
367
      pList = NULL;
×
368
      terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
369
      break;
×
370
    }
371
    
372
    SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr;
291,255,980✔
373
    bi.slotId = pColNode->slotId;
291,277,251✔
374
    void* tmp = taosArrayPush(pList, &bi);
291,285,995✔
375
    if (!tmp) {
291,285,995!
376
      taosArrayDestroy(pList);
×
377
      pList = NULL;
×
378
      break;
×
379
    }
380
  }
381

382
  return pList;
239,392,890✔
383
}
384

385
SSDataBlock* createDataBlockFromDescNode(void* p) {
2,147,483,647✔
386
  SDataBlockDescNode* pNode = (SDataBlockDescNode*)p;
2,147,483,647✔
387
  int32_t      numOfCols = LIST_LENGTH(pNode->pSlots);
2,147,483,647!
388
  SSDataBlock* pBlock = NULL;
2,147,483,647✔
389
  int32_t      code = createDataBlock(&pBlock);
2,147,483,647✔
390
  if (code) {
2,147,483,647!
391
    terrno = code;
×
392
    return NULL;
×
393
  }
394

395
  pBlock->info.id.blockId = pNode->dataBlockId;
2,147,483,647✔
396
  pBlock->info.type = STREAM_INVALID;
2,147,483,647✔
397
  pBlock->info.calWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
2,147,483,647✔
398
  pBlock->info.watermark = INT64_MIN;
2,147,483,647✔
399

400
  for (int32_t i = 0; i < numOfCols; ++i) {
2,147,483,647✔
401
    SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
2,147,483,647✔
402
    if (!pDescNode) {
2,147,483,647!
403
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
404
      blockDataDestroy(pBlock);
×
405
      pBlock = NULL;
×
406
      terrno = TSDB_CODE_INVALID_PARA;
×
407
      break;
×
408
    }
409
    SColumnInfoData idata =
2,147,483,647✔
410
        createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);
2,147,483,647✔
411
    idata.info.scale = pDescNode->dataType.scale;
2,147,483,647✔
412
    idata.info.precision = pDescNode->dataType.precision;
2,147,483,647✔
413
    idata.info.noData = pDescNode->reserve;
2,147,483,647!
414

415
    code = blockDataAppendColInfo(pBlock, &idata);
2,147,483,647✔
416
    if (code != TSDB_CODE_SUCCESS) {
2,147,483,647✔
417
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
376,213!
418
      blockDataDestroy(pBlock);
376,213✔
419
      pBlock = NULL;
×
420
      terrno = code;
×
421
      break;
×
422
    }
423
  }
424

425
  return pBlock;
2,147,483,647✔
426
}
427

428
int32_t prepareDataBlockBuf(SSDataBlock* pDataBlock, SColMatchInfo* pMatchInfo) {
852,925,268✔
429
  SDataBlockInfo* pBlockInfo = &pDataBlock->info;
852,925,268✔
430

431
  for (int32_t i = 0; i < taosArrayGetSize(pMatchInfo->pList); ++i) {
2,147,483,647✔
432
    SColMatchItem* pItem = taosArrayGet(pMatchInfo->pList, i);
2,147,483,647✔
433
    if (!pItem) {
2,147,483,647!
434
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
435
      return terrno;
×
436
    }
437

438
    if (pItem->isPk) {
2,147,483,647!
439
      SColumnInfoData* pInfoData = taosArrayGet(pDataBlock->pDataBlock, pItem->dstSlotId);
41,987,723✔
440
      if (!pInfoData) {
41,808,316!
441
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
442
        return terrno;
×
443
      }
444
      pBlockInfo->pks[0].type = pInfoData->info.type;
41,808,316✔
445
      pBlockInfo->pks[1].type = pInfoData->info.type;
41,836,222✔
446

447
      // allocate enough buffer size, which is pInfoData->info.bytes
448
      if (IS_VAR_DATA_TYPE(pItem->dataType.type)) {
41,841,963!
449
        pBlockInfo->pks[0].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
13,424,033!
450
        if (pBlockInfo->pks[0].pData == NULL) {
13,396,988!
451
          return terrno;
×
452
        }
453

454
        pBlockInfo->pks[1].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
13,399,097!
455
        if (pBlockInfo->pks[1].pData == NULL) {
13,399,650!
456
          taosMemoryFreeClear(pBlockInfo->pks[0].pData);
×
457
          return terrno;
×
458
        }
459

460
        pBlockInfo->pks[0].nData = pInfoData->info.bytes;
13,398,034✔
461
        pBlockInfo->pks[1].nData = pInfoData->info.bytes;
13,400,096✔
462
      }
463

464
      break;
41,823,452✔
465
    }
466
  }
467

468
  return TSDB_CODE_SUCCESS;
853,078,975✔
469
}
470

471
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
6,655,103✔
472
  STransTagExprCtx* pCtx = pContext;
6,655,103✔
473
  SMetaReader*      mr = pCtx->pReader;
6,655,103✔
474
  bool              isTagCol = false, isTbname = false;
6,655,103✔
475
  if (nodeType(*pNode) == QUERY_NODE_COLUMN) {
6,655,103✔
476
    SColumnNode* pCol = (SColumnNode*)*pNode;
1,901,458✔
477
    if (pCol->colType == COLUMN_TYPE_TBNAME)
1,901,458!
478
      isTbname = true;
×
479
    else
480
      isTagCol = true;
1,901,458✔
481
  } else if (nodeType(*pNode) == QUERY_NODE_FUNCTION) {
4,753,645!
482
    SFunctionNode* pFunc = (SFunctionNode*)*pNode;
×
483
    if (pFunc->funcType == FUNCTION_TYPE_TBNAME) isTbname = true;
×
484
  }
485
  if (isTagCol) {
6,655,103✔
486
    SColumnNode* pSColumnNode = *(SColumnNode**)pNode;
1,901,458✔
487

488
    SValueNode* res = NULL;
1,901,458✔
489
    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
1,901,458✔
490
    if (NULL == res) {
1,901,458!
491
      return DEAL_RES_ERROR;
×
492
    }
493

494
    res->translate = true;
1,901,458✔
495
    res->node.resType = pSColumnNode->node.resType;
1,901,458✔
496

497
    STagVal tagVal = {0};
1,901,458✔
498
    tagVal.cid = pSColumnNode->colId;
1,901,458✔
499
    const char* p = mr->pAPI->extractTagVal(mr->me.ctbEntry.pTags, pSColumnNode->node.resType.type, &tagVal);
1,901,458✔
500
    if (p == NULL) {
1,901,458!
501
      res->node.resType.type = TSDB_DATA_TYPE_NULL;
×
502
    } else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {
1,901,458!
503
      int32_t len = ((const STag*)p)->len;
×
504
      res->datum.p = taosMemoryCalloc(len + 1, 1);
×
505
      if (NULL == res->datum.p) {
×
506
        return DEAL_RES_ERROR;
×
507
      }
508
      memcpy(res->datum.p, p, len);
×
509
    } else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) {
1,901,458!
510
      if (IS_STR_DATA_BLOB(pSColumnNode->node.resType.type)) {
1,901,458!
511
        return TSDB_CODE_BLOB_NOT_SUPPORT_TAG;
×
512
      }
513

514
      res->datum.p = taosMemoryCalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1, 1);
1,901,458!
515
      if (NULL == res->datum.p) {
1,901,458!
516
        return DEAL_RES_ERROR;
×
517
      }
518

519
      if (IS_STR_DATA_BLOB(pSColumnNode->node.resType.type)) {
1,901,458!
520
        memcpy(blobDataVal(res->datum.p), tagVal.pData, tagVal.nData);
×
521
        blobDataSetLen(res->datum.p, tagVal.nData);
×
522
      } else {
523
        memcpy(varDataVal(res->datum.p), tagVal.pData, tagVal.nData);
1,901,458!
524
        varDataSetLen(res->datum.p, tagVal.nData);
1,901,458✔
525
      }
526
    } else {
527
      int32_t code = nodesSetValueNodeValue(res, &(tagVal.i64));
×
528
      if (code != TSDB_CODE_SUCCESS) {
×
529
        return DEAL_RES_ERROR;
×
530
      }
531
    }
532
    nodesDestroyNode(*pNode);
1,901,458✔
533
    *pNode = (SNode*)res;
1,901,458✔
534
  } else if (isTbname) {
4,753,645!
535
    SValueNode* res = NULL;
×
536
    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
×
537
    if (NULL == res) {
×
538
      return DEAL_RES_ERROR;
×
539
    }
540

541
    res->translate = true;
×
542
    res->node.resType = ((SExprNode*)(*pNode))->resType;
×
543

544
    int32_t len = strlen(mr->me.name);
×
545
    res->datum.p = taosMemoryCalloc(len + VARSTR_HEADER_SIZE + 1, 1);
×
546
    if (NULL == res->datum.p) {
×
547
      return DEAL_RES_ERROR;
×
548
    }
549
    memcpy(varDataVal(res->datum.p), mr->me.name, len);
×
550
    varDataSetLen(res->datum.p, len);
×
551
    nodesDestroyNode(*pNode);
×
552
    *pNode = (SNode*)res;
×
553
  }
554

555
  return DEAL_RES_CONTINUE;
6,655,103✔
556
}
557

558
int32_t isQualifiedTable(STableKeyInfo* info, SNode* pTagCond, void* metaHandle, bool* pQualified, SStorageAPI* pAPI) {
950,729✔
559
  int32_t     code = TSDB_CODE_SUCCESS;
950,729✔
560
  SMetaReader mr = {0};
950,729✔
561

562
  pAPI->metaReaderFn.initReader(&mr, metaHandle, META_READER_LOCK, &pAPI->metaFn);
950,729✔
563
  code = pAPI->metaReaderFn.getEntryGetUidCache(&mr, info->uid);
950,729✔
564
  if (TSDB_CODE_SUCCESS != code) {
950,729!
565
    pAPI->metaReaderFn.clearReader(&mr);
×
566
    *pQualified = false;
×
567

568
    return TSDB_CODE_SUCCESS;
×
569
  }
570

571
  SNode* pTagCondTmp = NULL;
950,729✔
572
  code = nodesCloneNode(pTagCond, &pTagCondTmp);
950,729✔
573
  if (TSDB_CODE_SUCCESS != code) {
950,729!
574
    *pQualified = false;
×
575
    pAPI->metaReaderFn.clearReader(&mr);
×
576
    return code;
×
577
  }
578
  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
950,729✔
579
  nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &ctx);
950,729✔
580
  pAPI->metaReaderFn.clearReader(&mr);
950,729✔
581
  if (TSDB_CODE_SUCCESS != ctx.code) {
950,729!
582
    *pQualified = false;
×
583
    terrno = code;
×
584
    return code;
×
585
  }
586

587
  SNode* pNew = NULL;
950,729✔
588
  code = scalarCalculateConstants(pTagCondTmp, &pNew);
950,729✔
589
  if (TSDB_CODE_SUCCESS != code) {
950,729!
590
    terrno = code;
×
591
    nodesDestroyNode(pTagCondTmp);
×
592
    *pQualified = false;
×
593

594
    return code;
×
595
  }
596

597
  SValueNode* pValue = (SValueNode*)pNew;
950,729✔
598
  *pQualified = pValue->datum.b;
950,729!
599

600
  nodesDestroyNode(pNew);
950,729✔
601
  return TSDB_CODE_SUCCESS;
950,729✔
602
}
603

604
static EDealRes getColumn(SNode** pNode, void* pContext) {
294,127,561✔
605
  tagFilterAssist* pData = (tagFilterAssist*)pContext;
294,127,561✔
606
  SColumnNode*     pSColumnNode = NULL;
294,127,561✔
607
  if (QUERY_NODE_COLUMN == nodeType((*pNode))) {
294,165,590✔
608
    pSColumnNode = *(SColumnNode**)pNode;
104,722,927✔
609
  } else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {
189,547,486✔
610
    SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);
3,554,188✔
611
    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {
3,557,693✔
612
      pData->code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pSColumnNode);
3,446,942✔
613
      if (NULL == pSColumnNode) {
3,451,827!
614
        return DEAL_RES_ERROR;
×
615
      }
616
      pSColumnNode->colId = -1;
3,451,827✔
617
      pSColumnNode->colType = COLUMN_TYPE_TBNAME;
3,450,447✔
618
      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;
3,450,447✔
619
      pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;
3,449,678✔
620
      nodesDestroyNode(*pNode);
3,448,679✔
621
      *pNode = (SNode*)pSColumnNode;
3,450,447✔
622
    } else {
623
      return DEAL_RES_CONTINUE;
109,906✔
624
    }
625
  } else {
626
    return DEAL_RES_CONTINUE;
185,901,014✔
627
  }
628

629
  void* data = taosHashGet(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));
108,165,760✔
630
  if (!data) {
108,126,685✔
631
    int32_t tempRes =
632
        taosHashPut(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));
90,017,992✔
633
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
90,079,871!
634
      return DEAL_RES_ERROR;
×
635
    }
636
    pSColumnNode->slotId = pData->index++;
90,079,871✔
637
    SColumnInfo cInfo = {.colId = pSColumnNode->colId,
90,030,621✔
638
                         .type = pSColumnNode->node.resType.type,
90,046,377✔
639
                         .bytes = pSColumnNode->node.resType.bytes,
90,014,855✔
640
                         .pk = pSColumnNode->isPk};
90,035,974!
641
#if TAG_FILTER_DEBUG
642
    qDebug("tagfilter build column info, slotId:%d, colId:%d, type:%d", pSColumnNode->slotId, cInfo.colId, cInfo.type);
643
#endif
644
    void* tmp = taosArrayPush(pData->cInfoList, &cInfo);
90,014,223✔
645
    if (!tmp) {
90,057,395!
646
      return DEAL_RES_ERROR;
×
647
    }
648
  } else {
649
    SColumnNode* col = *(SColumnNode**)data;
18,108,693✔
650
    pSColumnNode->slotId = col->slotId;
18,108,702✔
651
  }
652

653
  return DEAL_RES_CONTINUE;
108,147,287✔
654
}
655

656
static int32_t createResultData(SDataType* pType, int32_t numOfRows, SScalarParam* pParam) {
86,777,074✔
657
  SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
86,777,074!
658
  if (pColumnData == NULL) {
86,768,129!
659
    return terrno;
×
660
  }
661

662
  pColumnData->info.type = pType->type;
86,768,129✔
663
  pColumnData->info.bytes = pType->bytes;
86,789,569✔
664
  pColumnData->info.scale = pType->scale;
86,756,038✔
665
  pColumnData->info.precision = pType->precision;
86,774,212✔
666

667
  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);
86,737,750✔
668
  if (code != TSDB_CODE_SUCCESS) {
86,751,760!
669
    terrno = code;
×
670
    releaseColInfoData(pColumnData);
×
671
    return terrno;
×
672
  }
673

674
  pParam->columnData = pColumnData;
86,751,760✔
675
  pParam->colAlloced = true;
86,752,405✔
676
  return TSDB_CODE_SUCCESS;
86,719,074✔
677
}
678

679
static void releaseColInfoData(void* pCol) {
29,791,339✔
680
  if (pCol) {
29,791,339!
681
    SColumnInfoData* col = (SColumnInfoData*)pCol;
29,792,611✔
682
    colDataDestroy(col);
29,792,611✔
683
    taosMemoryFree(col);
29,791,323!
684
  }
685
}
29,787,037✔
686

687
void freeItem(void* p) {
694,702,843✔
688
  STUidTagInfo* pInfo = p;
694,702,843✔
689
  if (pInfo->pTagVal != NULL) {
694,702,843✔
690
    taosMemoryFree(pInfo->pTagVal);
693,749,468!
691
  }
692
}
694,634,191✔
693

694
static int32_t genTagFilterDigest(const SNode* pTagCond, T_MD5_CTX* pContext) {
×
695
  if (pTagCond == NULL) {
×
696
    return TSDB_CODE_SUCCESS;
×
697
  }
698

699
  char*   payload = NULL;
×
700
  int32_t len = 0;
×
701
  int32_t code = nodesNodeToMsg(pTagCond, &payload, &len);
×
702
  if (code != TSDB_CODE_SUCCESS) {
×
703
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
704
    return code;
×
705
  }
706

707
  tMD5Init(pContext);
×
708
  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
×
709
  tMD5Final(pContext);
×
710

711
  taosMemoryFree(payload);
×
712
  return TSDB_CODE_SUCCESS;
×
713
}
714

715
static int32_t genTbGroupDigest(const SNode* pGroup, uint8_t* filterDigest, T_MD5_CTX* pContext) {
×
716
  int32_t code = TSDB_CODE_SUCCESS;
×
717
  int32_t lino = 0;
×
718
  char*   payload = NULL;
×
719
  int32_t len = 0;
×
720
  code = nodesNodeToMsg(pGroup, &payload, &len);
×
721
  QUERY_CHECK_CODE(code, lino, _end);
×
722

723
  if (filterDigest[0]) {
×
724
    payload = taosMemoryRealloc(payload, len + tListLen(pContext->digest));
×
725
    QUERY_CHECK_NULL(payload, code, lino, _end, terrno);
×
726
    memcpy(payload + len, filterDigest + 1, tListLen(pContext->digest));
×
727
    len += tListLen(pContext->digest);
×
728
  }
729

730
  tMD5Init(pContext);
×
731
  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
×
732
  tMD5Final(pContext);
×
733

734
_end:
×
735
  taosMemoryFree(payload);
×
736
  if (code != TSDB_CODE_SUCCESS) {
×
737
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
738
  }
739
  return code;
×
740
}
741

742
int32_t qGetColumnsFromNodeList(void* data, bool isList, SArray** pColList) {
85,401,771✔
743
  int32_t code = TSDB_CODE_SUCCESS;
85,401,771✔
744
  tagFilterAssist ctx = {0};
85,401,771✔
745
  ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
85,415,308✔
746
  if (ctx.colHash == NULL) {
85,399,700!
747
    code = terrno;
×
748
    goto end;
×
749
  }
750

751
  ctx.index = 0;
85,399,700✔
752
  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
85,399,700✔
753
  if (ctx.cInfoList == NULL) {
85,417,531✔
754
    code = terrno;
29,880✔
755
    goto end;
×
756
  }
757

758
  if (isList) {
85,387,651✔
759
    SNode* pNode = NULL;
28,410,755✔
760
    FOREACH(pNode, (SNodeList*)data) {
58,200,849!
761
      nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);
29,787,363✔
762
      if (TSDB_CODE_SUCCESS != ctx.code) {
29,789,376!
763
        code = ctx.code;
×
764
        goto end;
×
765
      }
766
      REPLACE_NODE(pNode);
29,789,376✔
767
    }
768
  } else {
769
    SNode* pNode = (SNode*)data;
56,976,896✔
770
    nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);
56,977,028✔
771
    if (TSDB_CODE_SUCCESS != ctx.code) {
56,991,755!
772
      code = ctx.code;
×
773
      goto end;
×
774
    }
775
  }
776
  
777
  if (pColList != NULL) *pColList = ctx.cInfoList;
85,385,632!
778
  ctx.cInfoList = NULL;
85,345,832✔
779

780
end:
85,365,448✔
781
  taosHashCleanup(ctx.colHash);
85,385,737✔
782
  taosArrayDestroy(ctx.cInfoList);
85,285,143✔
783
  return code;
85,309,024✔
784
}
785

786
static int32_t buildGroupInfo(SColumnInfoData* pValue, int32_t i, SArray* gInfo) {
17,305,901✔
787
  int32_t code = TSDB_CODE_SUCCESS;
17,305,901✔
788
  SStreamGroupValue* v = taosArrayReserve(gInfo, 1);
17,305,901✔
789
  if (v == NULL) {
17,305,901!
790
    code = terrno;
×
791
    goto end;
×
792
  }
793
  if (colDataIsNull_s(pValue, i)) {
34,611,802!
794
    v->isNull = true;
×
795
  } else {
796
    v->isNull = false;
17,305,901✔
797
    char* data = colDataGetData(pValue, i);
17,305,901!
798
    if (pValue->info.type == TSDB_DATA_TYPE_JSON) {
17,304,558!
799
      if (tTagIsJson(data)) {
×
800
        code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
×
801
        goto end;
×
802
      }
803
      if (tTagIsJsonNull(data)) {
×
804
        v->isNull = true;
×
805
        goto end;
×
806
      }
807
      int32_t len = getJsonValueLen(data);
×
808
      v->data.type = pValue->info.type;
×
809
      v->data.nData = len;
×
810
      v->data.pData = taosMemoryCalloc(1, len + 1);
×
811
      if (v->data.pData == NULL) {
×
812
        code = terrno;
×
813
        goto end;
×
814
      }
815
      memcpy(v->data.pData, data, len);
×
816
      qDebug("buildGroupInfo:%d add json data len:%d, data:%s", i, len, (char*)v->data.pData);
×
817
    } else if (IS_VAR_DATA_TYPE(pValue->info.type)) {
17,305,901!
818
      if (varDataTLen(data) > pValue->info.bytes) {
16,389,969!
819
        code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
×
820
        goto end;
×
821
      }
822
      v->data.type = pValue->info.type;
16,389,969✔
823
      v->data.nData = varDataLen(data);
16,389,969✔
824
      v->data.pData = taosMemoryCalloc(1, varDataLen(data) + 1);
16,389,969!
825
      if (v->data.pData == NULL) {
16,389,969!
826
        code = terrno;
×
827
        goto end;
×
828
      }
829
      memcpy(v->data.pData, varDataVal(data), varDataLen(data));
16,389,969!
830
      qDebug("buildGroupInfo:%d add var data type:%d, len:%d, data:%s", i, pValue->info.type, varDataLen(data), (char*)v->data.pData);
16,389,969✔
831
    } else if (pValue->info.type == TSDB_DATA_TYPE_DECIMAL) {  // reader todo decimal
915,932!
832
      v->data.type = pValue->info.type;
×
833
      v->data.nData = pValue->info.bytes;
×
834
      v->data.pData = taosMemoryCalloc(1, pValue->info.bytes);
×
835
      if (v->data.pData == NULL) {
×
836
        code = terrno;
×
837
        goto end;
×
838
      }
839
      memcpy(&v->data.pData, data, pValue->info.bytes);
×
840
      qDebug("buildGroupInfo:%d add data type:%d, data:%"PRId64, i, pValue->info.type, v->data.val);
×
841
    } else {  // reader todo decimal
842
      v->data.type = pValue->info.type;
915,932✔
843
      memcpy(&v->data.val, data, pValue->info.bytes);
915,932!
844
      qDebug("buildGroupInfo:%d add data type:%d, data:%"PRId64, i, pValue->info.type, v->data.val);
915,932✔
845
    }
846
  }
847
end:
696,127✔
848
  if (code != TSDB_CODE_SUCCESS) {
17,305,901!
849
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
850
    v->isNull = true;
×
851
  }
852
  return code;
17,305,901✔
853
}
854

855
static void getColInfoResultForGroupbyForStream(void* pVnode, SNodeList* group, STableListInfo* pTableListInfo,
1,907,725✔
856
                                   SStorageAPI* pAPI, SHashObj* groupIdMap) {
857
  int32_t      code = TSDB_CODE_SUCCESS;
1,907,725✔
858
  int32_t      lino = 0;
1,907,725✔
859
  SArray*      pBlockList = NULL;
1,907,725✔
860
  SSDataBlock* pResBlock = NULL;
1,907,725✔
861
  SArray*      groupData = NULL;
1,907,725✔
862
  SArray*      pUidTagList = NULL;
1,907,725✔
863
  SArray*      gInfo = NULL;
1,907,725✔
864
  int32_t      tbNameIndex = 0;
1,906,382✔
865

866
  int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);
1,906,382✔
867
  if (rows == 0) {
1,907,725!
868
    return;
×
869
  }
870

871
  pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));
1,907,725✔
872
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
1,907,725!
873

874
  for (int32_t i = 0; i < rows; ++i) {
14,182,296✔
875
    STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
12,274,571✔
876
    QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);
12,274,571!
877
    STUidTagInfo info = {.uid = pkeyInfo->uid};
12,274,571✔
878
    void*        tmp = taosArrayPush(pUidTagList, &info);
12,274,571✔
879
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
12,274,571!
880
  }
881

882
  code = pAPI->metaFn.getTableTags(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
1,907,725✔
883
  if (code != TSDB_CODE_SUCCESS) {
1,907,725!
884
    goto end;
×
885
  }
886

887
  SArray* pColList = NULL;
1,907,725✔
888
  code = qGetColumnsFromNodeList(group, true, &pColList);
1,907,725✔
889
  if (code != TSDB_CODE_SUCCESS) {
1,907,725!
890
    goto end;
×
891
  }
892

893
  for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {
4,681,462✔
894
    SColumnInfo* tmp = (SColumnInfo*)taosArrayGet(pColList, i);
2,773,737✔
895
    if (tmp != NULL && tmp->colId == -1) {
2,773,737!
896
      tbNameIndex = i;
1,907,725✔
897
    }
898
  }
899
  
900
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
1,907,725✔
901
  pResBlock = createTagValBlockForFilter(pColList, numOfTables, pUidTagList, pVnode, pAPI);
1,907,725✔
902
  taosArrayDestroy(pColList);
1,907,725✔
903
  if (pResBlock == NULL) {
1,907,725!
904
    code = terrno;
×
905
    goto end;
×
906
  }
907

908
  pBlockList = taosArrayInit(2, POINTER_BYTES);
1,907,725✔
909
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
1,907,725!
910

911
  void* tmp = taosArrayPush(pBlockList, &pResBlock);
1,907,725✔
912
  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
1,907,725!
913

914
  groupData = taosArrayInit(2, POINTER_BYTES);
1,907,725✔
915
  QUERY_CHECK_NULL(groupData, code, lino, end, terrno);
1,907,725!
916

917
  SNode* pNode = NULL;
1,907,725✔
918
  FOREACH(pNode, group) {
4,681,462!
919
    SScalarParam output = {0};
2,772,407✔
920

921
    switch (nodeType(pNode)) {
2,772,407!
922
      case QUERY_NODE_VALUE:
×
923
        break;
×
924
      case QUERY_NODE_COLUMN:
2,773,737✔
925
      case QUERY_NODE_OPERATOR:
926
      case QUERY_NODE_FUNCTION: {
927
        SExprNode* expNode = (SExprNode*)pNode;
2,773,737✔
928
        code = createResultData(&expNode->resType, rows, &output);
2,773,737✔
929
        if (code != TSDB_CODE_SUCCESS) {
2,773,737!
930
          goto end;
×
931
        }
932
        break;
2,773,737✔
933
      }
934

935
      default:
×
936
        code = TSDB_CODE_OPS_NOT_SUPPORT;
×
937
        goto end;
×
938
    }
939

940
    if (nodeType(pNode) == QUERY_NODE_COLUMN) {
2,773,737!
941
      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;
2,773,737✔
942
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);
2,773,737✔
943
      QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);
2,773,737!
944
      code = colDataAssign(output.columnData, pColInfo, rows, NULL);
2,773,737✔
UNCOV
945
    } else if (nodeType(pNode) == QUERY_NODE_VALUE) {
×
946
      continue;
×
947
    } else {
948
      code = scalarCalculate(pNode, pBlockList, &output, NULL, NULL);
×
949
    }
950

951
    if (code != TSDB_CODE_SUCCESS) {
2,773,737!
952
      releaseColInfoData(output.columnData);
×
953
      goto end;
×
954
    }
955

956
    void* tmp = taosArrayPush(groupData, &output.columnData);
2,773,737✔
957
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
2,773,737!
958
  }
959

960
  for (int i = 0; i < rows; i++) {
14,182,296✔
961
    gInfo = taosArrayInit(taosArrayGetSize(groupData), sizeof(SStreamGroupValue));
12,274,571✔
962
    QUERY_CHECK_NULL(gInfo, code, lino, end, terrno);
12,273,296!
963

964
    STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
12,273,296✔
965
    QUERY_CHECK_NULL(info, code, lino, end, terrno);
12,273,296!
966

967
    for (int j = 0; j < taosArrayGetSize(groupData); j++) {
29,479,906✔
968
      SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);
17,206,610✔
969
        int32_t ret = buildGroupInfo(pValue, i, gInfo);
17,206,610✔
970
        if (ret != TSDB_CODE_SUCCESS) {
17,206,610!
971
          qError("buildGroupInfo failed at line %d since %s", __LINE__, tstrerror(ret));
×
972
          goto end;
×
973
        }
974
        if (j == tbNameIndex) {
17,206,610✔
975
          SStreamGroupValue* v = taosArrayGetLast(gInfo);
12,273,296✔
976
          if (v != NULL){
12,274,571!
977
            v->isTbname = true;
12,274,571✔
978
            v->uid = info->uid;
12,274,571✔
979
          }
980
        }
981
    }
982

983
    int32_t ret = taosHashPut(groupIdMap, &info->uid, sizeof(info->uid), &gInfo, POINTER_BYTES);
12,273,296✔
984
    if (ret != TSDB_CODE_SUCCESS) {
12,273,296!
985
      qError("put groupid to map failed at line %d since %s", __LINE__, tstrerror(ret));
×
986
      goto end;
×
987
    }
988
    qDebug("put groupid to map gid:%" PRIu64, info->uid);
12,273,296✔
989
    gInfo = NULL;
12,274,571✔
990
  }
991

992
end:
1,907,725✔
993
  blockDataDestroy(pResBlock);
1,907,725✔
994
  taosArrayDestroy(pBlockList);
1,907,725✔
995
  taosArrayDestroyEx(pUidTagList, freeItem);
1,907,725✔
996
  taosArrayDestroyP(groupData, releaseColInfoData);
1,907,725✔
997
  taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
1,907,725✔
998

999
  if (code != TSDB_CODE_SUCCESS) {
1,907,725!
1000
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1001
  }
1002
}
1003

1004
int32_t getColInfoResultForGroupby(void* pVnode, SNodeList* group, STableListInfo* pTableListInfo, uint8_t* digest,
26,505,466✔
1005
                                   SStorageAPI* pAPI, bool initRemainGroups, SHashObj* groupIdMap) {
1006
  int32_t      code = TSDB_CODE_SUCCESS;
26,505,466✔
1007
  int32_t      lino = 0;
26,505,466✔
1008
  SArray*      pBlockList = NULL;
26,505,466✔
1009
  SSDataBlock* pResBlock = NULL;
26,505,466✔
1010
  void*        keyBuf = NULL;
26,505,557✔
1011
  SArray*      groupData = NULL;
26,505,557✔
1012
  SArray*      pUidTagList = NULL;
26,505,557✔
1013
  SArray*      tableList = NULL;
26,505,557✔
1014
  SArray*      gInfo = NULL;
26,505,557✔
1015

1016
  int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);
26,505,557✔
1017
  if (rows == 0) {
26,506,202!
1018
    return TSDB_CODE_SUCCESS;
×
1019
  } 
1020

1021
  T_MD5_CTX context = {0};
26,506,202✔
1022
  if (tsTagFilterCache && groupIdMap == NULL) {
26,506,202!
1023
    SNodeListNode* listNode = NULL;
×
1024
    code = nodesMakeNode(QUERY_NODE_NODE_LIST, (SNode**)&listNode);
×
1025
    if (TSDB_CODE_SUCCESS != code) {
×
1026
      goto end;
×
1027
    }
1028
    listNode->pNodeList = group;
×
1029
    code = genTbGroupDigest((SNode*)listNode, digest, &context);
×
1030
    QUERY_CHECK_CODE(code, lino, end);
×
1031

1032
    nodesFree(listNode);
×
1033

1034
    code = pAPI->metaFn.metaGetCachedTbGroup(pVnode, pTableListInfo->idInfo.suid, context.digest,
×
1035
                                             tListLen(context.digest), &tableList);
1036
    QUERY_CHECK_CODE(code, lino, end);
×
1037

1038
    if (tableList) {
×
1039
      taosArrayDestroy(pTableListInfo->pTableList);
×
1040
      pTableListInfo->pTableList = tableList;
×
1041
      qDebug("retrieve tb group list from cache, numOfTables:%d",
×
1042
             (int32_t)taosArrayGetSize(pTableListInfo->pTableList));
1043
      goto end;
×
1044
    }
1045
  }
1046

1047
  pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));
26,506,202✔
1048
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
26,505,557!
1049

1050
  for (int32_t i = 0; i < rows; ++i) {
174,629,393✔
1051
    STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
148,120,272✔
1052
    QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);
148,120,917!
1053
    STUidTagInfo info = {.uid = pkeyInfo->uid};
148,120,917✔
1054
    void*        tmp = taosArrayPush(pUidTagList, &info);
148,125,480✔
1055
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
148,125,480!
1056
  }
1057

1058
  code = pAPI->metaFn.getTableTags(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
26,509,121✔
1059
  if (code != TSDB_CODE_SUCCESS) {
26,503,010!
1060
    goto end;
×
1061
  }
1062

1063
  SArray* pColList = NULL;
26,503,010✔
1064
  code = qGetColumnsFromNodeList(group, true, &pColList); 
26,503,010✔
1065
  if (code != TSDB_CODE_SUCCESS) {
26,495,628!
1066
    goto end;
×
1067
  }
1068

1069
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
26,495,628✔
1070
  pResBlock = createTagValBlockForFilter(pColList, numOfTables, pUidTagList, pVnode, pAPI);
26,498,791✔
1071
  taosArrayDestroy(pColList);
26,506,202✔
1072
  if (pResBlock == NULL) {
26,503,931!
1073
    code = terrno;
×
1074
    goto end;
×
1075
  }
1076

1077
  //  int64_t st1 = taosGetTimestampUs();
1078
  //  qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
1079

1080
  pBlockList = taosArrayInit(2, POINTER_BYTES);
26,503,931✔
1081
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
26,503,195!
1082

1083
  void* tmp = taosArrayPush(pBlockList, &pResBlock);
26,503,931✔
1084
  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
26,503,931!
1085

1086
  groupData = taosArrayInit(2, POINTER_BYTES);
26,503,931✔
1087
  QUERY_CHECK_NULL(groupData, code, lino, end, terrno);
26,505,203✔
1088

1089
  SNode* pNode = NULL;
26,504,930✔
1090
  FOREACH(pNode, group) {
53,515,888!
1091
    SScalarParam output = {0};
27,015,979✔
1092

1093
    switch (nodeType(pNode)) {
27,015,979!
1094
      case QUERY_NODE_VALUE:
×
1095
        break;
×
1096
      case QUERY_NODE_COLUMN:
27,018,874✔
1097
      case QUERY_NODE_OPERATOR:
1098
      case QUERY_NODE_FUNCTION: {
1099
        SExprNode* expNode = (SExprNode*)pNode;
27,018,874✔
1100
        code = createResultData(&expNode->resType, rows, &output);
27,018,874✔
1101
        if (code != TSDB_CODE_SUCCESS) {
27,015,670!
1102
          goto end;
×
1103
        }
1104
        break;
27,015,670✔
1105
      }
1106

1107
      default:
×
1108
        code = TSDB_CODE_OPS_NOT_SUPPORT;
×
1109
        goto end;
×
1110
    }
1111

1112
    if (nodeType(pNode) == QUERY_NODE_COLUMN) {
27,015,670✔
1113
      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;
26,952,522✔
1114
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);
26,952,522✔
1115
      QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);
26,948,524!
1116
      code = colDataAssign(output.columnData, pColInfo, rows, NULL);
26,948,524✔
1117
    } else if (nodeType(pNode) == QUERY_NODE_VALUE) {
64,792!
1118
      continue;
×
1119
    } else {
1120
      code = scalarCalculate(pNode, pBlockList, &output, NULL, NULL);
63,376✔
1121
    }
1122

1123
    if (code != TSDB_CODE_SUCCESS) {
27,011,159!
1124
      releaseColInfoData(output.columnData);
×
1125
      goto end;
×
1126
    }
1127

1128
    void* tmp = taosArrayPush(groupData, &output.columnData);
27,015,705✔
1129
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
27,015,705!
1130
  }
1131

1132
  int32_t keyLen = 0;
26,498,286✔
1133
  SNode*  node;
1134
  FOREACH(node, group) {
53,499,429!
1135
    SExprNode* pExpr = (SExprNode*)node;
27,011,239✔
1136
    keyLen += pExpr->resType.bytes;
27,011,239✔
1137
  }
1138

1139
  int32_t nullFlagSize = sizeof(int8_t) * LIST_LENGTH(group);
26,498,654!
1140
  keyLen += nullFlagSize;
26,498,021✔
1141

1142
  keyBuf = taosMemoryCalloc(1, keyLen);
26,498,021!
1143
  if (keyBuf == NULL) {
26,488,565!
1144
    code = terrno;
×
1145
    goto end;
×
1146
  }
1147

1148
  if (initRemainGroups) {
26,488,565✔
1149
    pTableListInfo->remainGroups =
12,173,013✔
1150
        taosHashInit(rows, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
12,164,945✔
1151
    if (pTableListInfo->remainGroups == NULL) {
12,173,013!
1152
      code = terrno;
×
1153
      goto end;
×
1154
    }
1155
  }
1156

1157
  for (int i = 0; i < rows; i++) {
174,497,212✔
1158
    STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
148,007,810✔
1159
    QUERY_CHECK_NULL(info, code, lino, end, terrno);
148,006,131!
1160

1161
    if (groupIdMap != NULL){
148,006,131✔
1162
      gInfo = taosArrayInit(taosArrayGetSize(groupData), sizeof(SStreamGroupValue));
84,730✔
1163
    }
1164
    
1165
    char* isNull = (char*)keyBuf;
148,004,446✔
1166
    char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(group);
148,004,446!
1167
    for (int j = 0; j < taosArrayGetSize(groupData); j++) {
300,850,740✔
1168
      SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);
152,844,293✔
1169

1170
      if (groupIdMap != NULL && gInfo != NULL) {
152,849,956!
1171
        int32_t ret = buildGroupInfo(pValue, i, gInfo);
98,016✔
1172
        if (ret != TSDB_CODE_SUCCESS) {
98,016!
1173
          qError("buildGroupInfo failed at line %d since %s", __LINE__, tstrerror(ret));
×
1174
          taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
×
1175
          gInfo = NULL;
×
1176
        }
1177
      }
1178
      
1179
      if (colDataIsNull_s(pValue, i)) {
305,698,200!
1180
        isNull[j] = 1;
282,918✔
1181
      } else {
1182
        isNull[j] = 0;
152,565,326✔
1183
        char* data = colDataGetData(pValue, i);
152,570,036!
1184
        if (pValue->info.type == TSDB_DATA_TYPE_JSON) {
152,566,981✔
1185
          if (tTagIsJson(data)) {
154,044✔
1186
            code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
7,002✔
1187
            goto end;
7,002✔
1188
          }
1189
          if (tTagIsJsonNull(data)) {
147,042!
1190
            isNull[j] = 1;
×
1191
            continue;
×
1192
          }
1193
          int32_t len = getJsonValueLen(data);
147,042✔
1194
          memcpy(pStart, data, len);
147,042!
1195
          pStart += len;
147,042✔
1196
        } else if (IS_VAR_DATA_TYPE(pValue->info.type)) {
152,416,438!
1197
          if (IS_STR_DATA_BLOB(pValue->info.type)) {
138,466,033!
1198
            if (blobDataTLen(data) > TSDB_MAX_BLOB_LEN) {
485!
1199
              code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
×
1200
              goto end;
×
1201
            }
1202
            memcpy(pStart, data, blobDataTLen(data));
×
1203
            pStart += blobDataTLen(data);
×
1204
          } else {
1205
            if (varDataTLen(data) > pValue->info.bytes) {
138,461,768✔
1206
              code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
771✔
1207
              goto end;
771✔
1208
            }
1209
            memcpy(pStart, data, varDataTLen(data));
138,465,644!
1210
            pStart += varDataTLen(data);
138,468,264✔
1211
          }
1212
        } else {
1213
          memcpy(pStart, data, pValue->info.bytes);
13,952,296!
1214
          pStart += pValue->info.bytes;
13,953,808✔
1215
        }
1216
      }
1217
    }
1218

1219
    int32_t len = (int32_t)(pStart - (char*)keyBuf);
147,977,618✔
1220
    info->groupId = calcGroupId(keyBuf, len);
147,977,618✔
1221
    if (groupIdMap != NULL && gInfo != NULL) {
148,002,356!
1222
      int32_t ret = taosHashPut(groupIdMap, &info->groupId, sizeof(info->groupId), &gInfo, POINTER_BYTES);
84,730✔
1223
      if (ret != TSDB_CODE_SUCCESS) {
84,730!
1224
        qError("put groupid to map failed at line %d since %s", __LINE__, tstrerror(ret));
×
1225
        taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
×
1226
      }
1227
      qDebug("put groupid to map gid:%" PRIu64, info->groupId);
84,730!
1228
      gInfo = NULL;
84,730✔
1229
    }
1230
    if (initRemainGroups) {
148,002,356✔
1231
      // groupId ~ table uid
1232
      code = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid),
70,823,560✔
1233
                         sizeof(info->uid));
1234
      if (code == TSDB_CODE_DUP_KEY) {
70,814,909✔
1235
        code = TSDB_CODE_SUCCESS;
1,828,999✔
1236
      }
1237
      QUERY_CHECK_CODE(code, lino, end);
70,814,909!
1238
    }
1239
  }
1240

1241
  if (tsTagFilterCache && groupIdMap == NULL) {
26,489,402!
1242
    tableList = taosArrayDup(pTableListInfo->pTableList, NULL);
×
1243
    QUERY_CHECK_NULL(tableList, code, lino, end, terrno);
×
1244

1245
    code = pAPI->metaFn.metaPutTbGroupToCache(pVnode, pTableListInfo->idInfo.suid, context.digest,
×
1246
                                              tListLen(context.digest), tableList,
1247
                                              taosArrayGetSize(tableList) * sizeof(STableKeyInfo));
×
1248
    QUERY_CHECK_CODE(code, lino, end);
×
1249
  }
1250

1251
  //  int64_t st2 = taosGetTimestampUs();
1252
  //  qDebug("calculate tag block rows:%d, cost:%ld us", rows, st2-st1);
1253

1254
end:
26,497,946✔
1255
  taosMemoryFreeClear(keyBuf);
26,505,593!
1256
  blockDataDestroy(pResBlock);
26,505,578✔
1257
  taosArrayDestroy(pBlockList);
26,502,957✔
1258
  taosArrayDestroyEx(pUidTagList, freeItem);
26,501,015✔
1259
  taosArrayDestroyP(groupData, releaseColInfoData);
26,504,830✔
1260
  taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
26,502,308✔
1261

1262
  if (code != TSDB_CODE_SUCCESS) {
26,502,308✔
1263
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
7,773!
1264
  }
1265
  return code;
26,503,697✔
1266
}
1267

1268
static int32_t nameComparFn(const void* p1, const void* p2) {
2,661,432✔
1269
  const char* pName1 = *(const char**)p1;
2,661,432✔
1270
  const char* pName2 = *(const char**)p2;
2,661,432✔
1271

1272
  int32_t ret = strcmp(pName1, pName2);
2,661,432!
1273
  if (ret == 0) {
2,661,432✔
1274
    return 0;
45,480✔
1275
  } else {
1276
    return (ret > 0) ? 1 : -1;
2,615,952✔
1277
  }
1278
}
1279

1280
static SArray* getTableNameList(const SNodeListNode* pList) {
1,080,373✔
1281
  int32_t    code = TSDB_CODE_SUCCESS;
1,080,373✔
1282
  int32_t    lino = 0;
1,080,373✔
1283
  int32_t    len = LIST_LENGTH(pList->pNodeList);
1,080,373!
1284
  SListCell* cell = pList->pNodeList->pHead;
1,080,373✔
1285

1286
  SArray* pTbList = taosArrayInit(len, POINTER_BYTES);
1,079,272✔
1287
  QUERY_CHECK_NULL(pTbList, code, lino, _end, terrno);
1,080,373!
1288

1289
  for (int i = 0; i < pList->pNodeList->length; i++) {
3,283,653✔
1290
    SValueNode* valueNode = (SValueNode*)cell->pNode;
2,203,280✔
1291
    if (!IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {
2,203,280!
1292
      terrno = TSDB_CODE_INVALID_PARA;
×
1293
      taosArrayDestroy(pTbList);
×
1294
      return NULL;
×
1295
    }
1296

1297
    char* name = varDataVal(valueNode->datum.p);
2,203,280✔
1298
    void* tmp = taosArrayPush(pTbList, &name);
2,203,280✔
1299
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
2,203,280!
1300
    cell = cell->pNext;
2,203,280✔
1301
  }
1302

1303
  size_t numOfTables = taosArrayGetSize(pTbList);
1,080,373✔
1304

1305
  // order the name
1306
  taosArraySort(pTbList, nameComparFn);
1,080,373✔
1307

1308
  // remove the duplicates
1309
  SArray* pNewList = taosArrayInit(taosArrayGetSize(pTbList), sizeof(void*));
1,080,373✔
1310
  QUERY_CHECK_NULL(pNewList, code, lino, _end, terrno);
1,080,373!
1311
  void* tmpTbl = taosArrayGet(pTbList, 0);
1,080,373✔
1312
  QUERY_CHECK_NULL(tmpTbl, code, lino, _end, terrno);
1,080,373!
1313
  void* tmp = taosArrayPush(pNewList, tmpTbl);
1,080,373✔
1314
  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
1,080,373!
1315

1316
  for (int32_t i = 1; i < numOfTables; ++i) {
2,203,280✔
1317
    char** name = taosArrayGetLast(pNewList);
1,122,907✔
1318
    char** nameInOldList = taosArrayGet(pTbList, i);
1,122,907✔
1319
    QUERY_CHECK_NULL(nameInOldList, code, lino, _end, terrno);
1,121,806!
1320
    if (strcmp(*name, *nameInOldList) == 0) {
1,121,806!
1321
      continue;
24,304✔
1322
    }
1323

1324
    tmp = taosArrayPush(pNewList, nameInOldList);
1,098,603✔
1325
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
1,098,603!
1326
  }
1327

1328
_end:
1,080,373✔
1329
  taosArrayDestroy(pTbList);
1,080,373✔
1330
  if (code != TSDB_CODE_SUCCESS) {
1,080,373!
1331
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1332
    return NULL;
×
1333
  }
1334
  return pNewList;
1,080,373✔
1335
}
1336

1337
static int tableUidCompare(const void* a, const void* b) {
×
1338
  uint64_t u1 = *(uint64_t*)a;
×
1339
  uint64_t u2 = *(uint64_t*)b;
×
1340

1341
  if (u1 == u2) {
×
1342
    return 0;
×
1343
  }
1344

1345
  return u1 < u2 ? -1 : 1;
×
1346
}
1347

1348
static int32_t filterTableInfoCompare(const void* a, const void* b) {
51,105,821✔
1349
  STUidTagInfo* p1 = (STUidTagInfo*)a;
51,105,821✔
1350
  STUidTagInfo* p2 = (STUidTagInfo*)b;
51,105,821✔
1351

1352
  if (p1->uid == p2->uid) {
51,105,821!
1353
    return 0;
×
1354
  }
1355

1356
  return p1->uid < p2->uid ? -1 : 1;
51,105,821✔
1357
}
1358

1359
static FilterCondType checkTagCond(SNode* cond) {
61,512,419✔
1360
  if (nodeType(cond) == QUERY_NODE_OPERATOR) {
61,512,419✔
1361
    return FILTER_NO_LOGIC;
40,608,496✔
1362
  }
1363
  if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
20,907,985!
1364
    return FILTER_AND;
3,539,751✔
1365
  }
1366
  return FILTER_OTHER;
17,351,779✔
1367
}
1368

1369
static int32_t optimizeTbnameInCond(void* pVnode, int64_t suid, SArray* list, SNode* cond, SStorageAPI* pAPI) {
61,507,499✔
1370
  int32_t ret = -1;
61,507,499✔
1371
  int32_t ntype = nodeType(cond);
61,507,499✔
1372

1373
  if (ntype == QUERY_NODE_OPERATOR) {
61,516,188✔
1374
    ret = optimizeTbnameInCondImpl(pVnode, list, cond, pAPI, suid);
40,611,977✔
1375
  }
1376

1377
  if (ntype != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
61,448,101✔
1378
    return ret;
44,119,651✔
1379
  }
1380

1381
  bool                 hasTbnameCond = false;
17,341,323✔
1382
  SLogicConditionNode* pNode = (SLogicConditionNode*)cond;
17,341,323✔
1383
  SNodeList*           pList = (SNodeList*)pNode->pParameterList;
17,341,323✔
1384

1385
  int32_t len = LIST_LENGTH(pList);
17,357,669!
1386
  if (len <= 0) {
17,359,259!
1387
    return ret;
×
1388
  }
1389

1390
  SListCell* cell = pList->pHead;
17,359,259✔
1391
  for (int i = 0; i < len; i++) {
52,960,766✔
1392
    if (cell == NULL) break;
35,674,690!
1393
    if (optimizeTbnameInCondImpl(pVnode, list, cell->pNode, pAPI, suid) == 0) {
35,674,690✔
1394
      hasTbnameCond = true;
39,323✔
1395
      break;
39,323✔
1396
    }
1397
    cell = cell->pNext;
35,606,644✔
1398
  }
1399

1400
  taosArraySort(list, filterTableInfoCompare);
17,325,399✔
1401
  taosArrayRemoveDuplicate(list, filterTableInfoCompare, NULL);
17,349,106✔
1402

1403
  if (hasTbnameCond) {
17,364,237✔
1404
    ret = pAPI->metaFn.getTableTagsByUid(pVnode, suid, list);
39,323✔
1405
  }
1406

1407
  return ret;
17,365,690✔
1408
}
1409

1410
// only return uid that does not contained in pExistedUidList
1411
static int32_t optimizeTbnameInCondImpl(void* pVnode, SArray* pExistedUidList, SNode* pTagCond, SStorageAPI* pStoreAPI,
76,282,804✔
1412
                                        uint64_t suid) {
1413
  if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {
76,282,804✔
1414
    return -1;
17,168✔
1415
  }
1416

1417
  SOperatorNode* pNode = (SOperatorNode*)pTagCond;
76,260,878✔
1418
  if (pNode->opType != OP_TYPE_IN) {
76,260,878✔
1419
    return -1;
71,935,053✔
1420
  }
1421

1422
  if ((pNode->pLeft != NULL && ((nodeType(pNode->pLeft) == QUERY_NODE_FUNCTION &&
4,316,061!
1423
                                 ((SFunctionNode*)pNode->pLeft)->funcType == FUNCTION_TYPE_TBNAME)) ||
1,080,373!
1424
       (nodeType(pNode->pLeft) == QUERY_NODE_COLUMN && ((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME)) &&
3,233,502!
1425
      (pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {
1,074,521!
1426
    SNodeListNode* pList = (SNodeListNode*)pNode->pRight;
1,080,373✔
1427

1428
    int32_t len = LIST_LENGTH(pList->pNodeList);
1,080,373!
1429
    if (len <= 0) {
1,080,373!
1430
      return -1;
×
1431
    }
1432

1433
    SArray*   pTbList = getTableNameList(pList);
1,080,373✔
1434
    int32_t   numOfTables = taosArrayGetSize(pTbList);
1,080,373✔
1435
    SHashObj* uHash = NULL;
1,080,373✔
1436

1437
    size_t numOfExisted = taosArrayGetSize(pExistedUidList);  // len > 0 means there already have uids
1,080,373✔
1438
    if (numOfExisted > 0) {
1,080,373✔
1439
      uHash = taosHashInit(numOfExisted / 0.7, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
6,016!
1440
      if (!uHash) {
6,016!
1441
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1442
        return terrno;
×
1443
      }
1444

1445
      for (int i = 0; i < numOfExisted; i++) {
6,012,992✔
1446
        STUidTagInfo* pTInfo = taosArrayGet(pExistedUidList, i);
6,006,976✔
1447
        if (!pTInfo) {
6,006,976!
1448
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1449
          return terrno;
×
1450
        }
1451
        int32_t tempRes = taosHashPut(uHash, &pTInfo->uid, sizeof(uint64_t), &i, sizeof(i));
6,006,976✔
1452
        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
6,006,976!
1453
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
1454
          return tempRes;
×
1455
        }
1456
      }
1457
    }
1458

1459
    for (int i = 0; i < numOfTables; i++) {
3,210,483✔
1460
      char* name = taosArrayGetP(pTbList, i);
2,154,672✔
1461

1462
      uint64_t uid = 0, csuid = 0;
2,154,672✔
1463
      if (pStoreAPI->metaFn.getTableUidByName(pVnode, name, &uid) == 0) {
2,153,571✔
1464
        ETableType tbType = TSDB_TABLE_MAX;
1,083,692✔
1465
        if (pStoreAPI->metaFn.getTableTypeSuidByName(pVnode, name, &tbType, &csuid) == 0 &&
1,083,692!
1466
            tbType == TSDB_CHILD_TABLE) {
1,083,692✔
1467
          if (suid != csuid) {
1,059,130✔
1468
            continue;
18,804✔
1469
          }
1470
          if (NULL == uHash || taosHashGet(uHash, &uid, sizeof(uid)) == NULL) {
1,040,326✔
1471
            STUidTagInfo s = {.uid = uid, .name = name, .pTagVal = NULL};
1,037,318✔
1472
            void*        tmp = taosArrayPush(pExistedUidList, &s);
1,037,318✔
1473
            if (!tmp) {
1,037,318!
1474
              return terrno;
×
1475
            }
1476
          }
1477
        } else {
1478
          taosArrayDestroy(pTbList);
24,562✔
1479
          taosHashCleanup(uHash);
24,562✔
1480
          return -1;
24,562✔
1481
        }
1482
      } else {
1483
        //        qWarn("failed to get tableIds from by table name: %s, reason: %s", name, tstrerror(terrno));
1484
        terrno = 0;
1,070,980✔
1485
      }
1486
    }
1487

1488
    taosHashCleanup(uHash);
1,055,811✔
1489
    taosArrayDestroy(pTbList);
1,055,811✔
1490
    return 0;
1,055,811✔
1491
  }
1492

1493
  return -1;
3,240,502✔
1494
}
1495

1496
SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, SArray* pUidTagList, void* pVnode,
96,164,851✔
1497
                                        SStorageAPI* pStorageAPI) {
1498
  int32_t      code = TSDB_CODE_SUCCESS;
96,164,851✔
1499
  int32_t      lino = 0;
96,164,851✔
1500
  SSDataBlock* pResBlock = NULL;
96,164,851✔
1501
  code = createDataBlock(&pResBlock);
96,181,955✔
1502
  QUERY_CHECK_CODE(code, lino, _end);
96,154,897!
1503

1504
  for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {
197,018,210✔
1505
    SColumnInfoData colInfo = {0};
100,857,323✔
1506
    void*           tmp = taosArrayGet(pColList, i);
100,809,038✔
1507
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
100,751,159!
1508
    colInfo.info = *(SColumnInfo*)tmp;
100,751,159✔
1509
    code = blockDataAppendColInfo(pResBlock, &colInfo);
100,721,605✔
1510
    QUERY_CHECK_CODE(code, lino, _end);
100,847,999!
1511
  }
1512

1513
  code = blockDataEnsureCapacity(pResBlock, numOfTables);
96,149,270✔
1514
  if (code != TSDB_CODE_SUCCESS) {
96,151,548!
1515
    terrno = code;
×
1516
    blockDataDestroy(pResBlock);
×
1517
    return NULL;
×
1518
  }
1519

1520
  pResBlock->info.rows = numOfTables;
96,151,548✔
1521

1522
  int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);
96,121,155✔
1523

1524
  for (int32_t i = 0; i < numOfTables; i++) {
807,182,320✔
1525
    STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);
710,980,415✔
1526
    QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
710,988,645✔
1527

1528
    for (int32_t j = 0; j < numOfCols; j++) {
1,453,353,556✔
1529
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, j);
742,144,707✔
1530
      QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
742,246,271!
1531

1532
      if (pColInfo->info.colId == -1) {  // tbname
742,246,271✔
1533
        char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
32,316,699✔
1534
        if (p1->name != NULL) {
32,322,373✔
1535
          STR_TO_VARSTR(str, p1->name);
1,034,836!
1536
        } else {  // name is not retrieved during filter
1537
          code = pStorageAPI->metaFn.getTableNameByUid(pVnode, p1->uid, str);
31,287,659✔
1538
          QUERY_CHECK_CODE(code, lino, _end);
31,285,284!
1539
        }
1540

1541
        code = colDataSetVal(pColInfo, i, str, false);
32,322,602✔
1542
        QUERY_CHECK_CODE(code, lino, _end);
32,322,306!
1543
#if TAG_FILTER_DEBUG
1544
        qDebug("tagfilter uid:%ld, tbname:%s", *uid, str + 2);
1545
#endif
1546
      } else {
1547
        STagVal tagVal = {0};
709,845,845✔
1548
        tagVal.cid = pColInfo->info.colId;
709,905,894✔
1549
        if (p1->pTagVal == NULL) {
709,890,734!
1550
          colDataSetNULL(pColInfo, i);
×
1551
        } else {
1552
          const char* p = pStorageAPI->metaFn.extractTagVal(p1->pTagVal, pColInfo->info.type, &tagVal);
709,864,456✔
1553

1554
          if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {
709,958,698✔
1555
            colDataSetNULL(pColInfo, i);
15,238,628!
1556
          } else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) {
694,718,289✔
1557
            code = colDataSetVal(pColInfo, i, p, false);
2,838,160✔
1558
            QUERY_CHECK_CODE(code, lino, _end);
2,838,160!
1559
          } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
1,126,084,241!
1560
            if (IS_STR_DATA_BLOB(pColInfo->info.type)) {
434,352,656✔
1561
              QUERY_CHECK_CODE(code = TSDB_CODE_BLOB_NOT_SUPPORT_TAG, lino, _end);
×
1562
            }
1563
            char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);
434,221,867!
1564
            QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
434,293,349!
1565
            varDataSetLen(tmp, tagVal.nData);
434,293,349✔
1566
            memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);
434,289,521!
1567
            code = colDataSetVal(pColInfo, i, tmp, false);
434,287,161✔
1568
#if TAG_FILTER_DEBUG
1569
            qDebug("tagfilter varch:%s", tmp + 2);
1570
#endif
1571
            taosMemoryFree(tmp);
434,321,541!
1572
            QUERY_CHECK_CODE(code, lino, _end);
434,288,733!
1573
          } else {
1574
            code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false);
257,539,305✔
1575
            QUERY_CHECK_CODE(code, lino, _end);
257,684,820!
1576
#if TAG_FILTER_DEBUG
1577
            if (pColInfo->info.type == TSDB_DATA_TYPE_INT) {
1578
              qDebug("tagfilter int:%d", *(int*)(&tagVal.i64));
1579
            } else if (pColInfo->info.type == TSDB_DATA_TYPE_DOUBLE) {
1580
              qDebug("tagfilter double:%f", *(double*)(&tagVal.i64));
1581
            }
1582
#endif
1583
          }
1584
        }
1585
      }
1586
    }
1587
  }
1588

1589
_end:
96,183,424✔
1590
  if (code != TSDB_CODE_SUCCESS) {
96,201,912✔
1591
    blockDataDestroy(pResBlock);
2,587✔
1592
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1593
    terrno = code;
×
1594
    return NULL;
×
1595
  }
1596
  return pResBlock;
96,199,436✔
1597
}
1598

1599
static int32_t doSetQualifiedUid(STableListInfo* pListInfo, SArray* pUidList, const SArray* pUidTagList,
56,918,850✔
1600
                                 bool* pResultList, bool addUid) {
1601
  taosArrayClear(pUidList);
56,918,850✔
1602

1603
  STableKeyInfo info = {.uid = 0, .groupId = 0};
56,920,597✔
1604
  int32_t       numOfTables = taosArrayGetSize(pUidTagList);
56,925,220✔
1605
  for (int32_t i = 0; i < numOfTables; ++i) {
591,380,516✔
1606
    if (pResultList[i]) {
534,388,682!
1607
      STUidTagInfo* tmpTag = (STUidTagInfo*)taosArrayGet(pUidTagList, i);
296,399,930✔
1608
      if (!tmpTag) {
296,403,254!
1609
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1610
        return terrno;
×
1611
      }
1612
      uint64_t uid = tmpTag->uid;
296,403,254✔
1613
      qDebug("tagfilter get uid:%" PRId64 ", res:%d", uid, pResultList[i]);
296,435,862!
1614

1615
      info.uid = uid;
296,489,052✔
1616
      void* p = taosArrayPush(pListInfo->pTableList, &info);
296,489,052✔
1617
      if (p == NULL) {
296,474,736!
1618
        return terrno;
×
1619
      }
1620

1621
      if (addUid) {
296,474,736!
1622
        void* tmp = taosArrayPush(pUidList, &uid);
×
1623
        if (tmp == NULL) {
×
1624
          return terrno;
×
1625
        }
1626
      }
1627
    }
1628
  }
1629

1630
  return TSDB_CODE_SUCCESS;
56,991,834✔
1631
}
1632

1633
static int32_t copyExistedUids(SArray* pUidTagList, const SArray* pUidList) {
61,512,797✔
1634
  int32_t code = TSDB_CODE_SUCCESS;
61,512,797✔
1635
  int32_t numOfExisted = taosArrayGetSize(pUidList);
61,512,797✔
1636
  if (numOfExisted == 0) {
61,488,778✔
1637
    return code;
52,832,235✔
1638
  }
1639

1640
  for (int32_t i = 0; i < numOfExisted; ++i) {
107,062,492✔
1641
    uint64_t* uid = taosArrayGet(pUidList, i);
98,405,949✔
1642
    if (!uid) {
98,404,558!
1643
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1644
      return terrno;
×
1645
    }
1646
    STUidTagInfo info = {.uid = *uid};
98,404,558✔
1647
    void*        tmp = taosArrayPush(pUidTagList, &info);
98,405,949✔
1648
    if (!tmp) {
98,405,949!
1649
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1650
      return code;
×
1651
    }
1652
  }
1653
  return code;
8,656,543✔
1654
}
1655

1656
static int32_t doFilterByTagCond(STableListInfo* pListInfo, SArray* pUidList, SNode* pTagCond, void* pVnode,
855,687,443✔
1657
                                 SIdxFltStatus status, SStorageAPI* pAPI, bool addUid, bool* listAdded, void* pStreamInfo) {
1658
  *listAdded = false;
855,687,443✔
1659
  if (pTagCond == NULL) {
855,748,318✔
1660
    return TSDB_CODE_SUCCESS;
794,123,081✔
1661
  }
1662

1663
  terrno = TSDB_CODE_SUCCESS;
61,625,237✔
1664

1665
  int32_t      lino = 0;
61,502,869✔
1666
  int32_t      code = TSDB_CODE_SUCCESS;
61,502,869✔
1667
  SArray*      pBlockList = NULL;
61,502,869✔
1668
  SSDataBlock* pResBlock = NULL;
61,502,869✔
1669
  SScalarParam output = {0};
61,483,258✔
1670
  SArray*      pUidTagList = NULL;
61,478,745✔
1671

1672
  SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};
61,478,745✔
1673

1674
  //  int64_t stt = taosGetTimestampUs();
1675
  pUidTagList = taosArrayInit(10, sizeof(STUidTagInfo));
61,483,454✔
1676
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
61,476,055!
1677

1678
  code = copyExistedUids(pUidTagList, pUidList);
61,476,055✔
1679
  QUERY_CHECK_CODE(code, lino, end);
61,511,506!
1680

1681
  FilterCondType condType = checkTagCond(pTagCond);
61,511,506✔
1682

1683
  int32_t filter = optimizeTbnameInCond(pVnode, pListInfo->idInfo.suid, pUidTagList, pTagCond, pAPI);
61,482,500✔
1684
  if (filter == 0) {  // tbname in filter is activated, do nothing and return
61,471,568✔
1685
    taosArrayClear(pUidList);
1,055,811✔
1686

1687
    int32_t numOfRows = taosArrayGetSize(pUidTagList);
1,055,811✔
1688
    code = taosArrayEnsureCap(pUidList, numOfRows);
1,055,811✔
1689
    QUERY_CHECK_CODE(code, lino, end);
1,055,811!
1690

1691
    for (int32_t i = 0; i < numOfRows; ++i) {
8,100,105✔
1692
      STUidTagInfo* pInfo = taosArrayGet(pUidTagList, i);
7,044,294✔
1693
      QUERY_CHECK_NULL(pInfo, code, lino, end, terrno);
7,044,294!
1694
      void* tmp = taosArrayPush(pUidList, &pInfo->uid);
7,044,294✔
1695
      QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
7,044,294!
1696
    }
1697
    terrno = 0;
1,055,811✔
1698
  } else {
1699
    if ((condType == FILTER_NO_LOGIC || condType == FILTER_AND) && status != SFLT_NOT_INDEX) {
60,415,757✔
1700
      code = pAPI->metaFn.getTableTagsByUid(pVnode, pListInfo->idInfo.suid, pUidTagList);
8,694,851✔
1701
    } else {
1702
      code = pAPI->metaFn.getTableTags(pVnode, pListInfo->idInfo.suid, pUidTagList);
51,720,906✔
1703
    }
1704
    if (code != TSDB_CODE_SUCCESS) {
60,452,056!
1705
      qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), pListInfo->idInfo.suid);
×
1706
      terrno = code;
×
1707
      QUERY_CHECK_CODE(code, lino, end);
×
1708
    }
1709
  }
1710

1711
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
61,507,867✔
1712
  if (numOfTables == 0) {
61,513,535✔
1713
    goto end;
4,521,762✔
1714
  }
1715

1716
  SArray* pColList = NULL;
56,991,773✔
1717
  code = qGetColumnsFromNodeList(pTagCond, false, &pColList); 
56,995,458✔
1718
  if (code != TSDB_CODE_SUCCESS) {
56,908,698!
1719
    goto end;
×
1720
  }
1721
  pResBlock = createTagValBlockForFilter(pColList, numOfTables, pUidTagList, pVnode, pAPI);
56,908,698✔
1722
  taosArrayDestroy(pColList);
56,978,698✔
1723
  if (pResBlock == NULL) {
56,969,555!
1724
    code = terrno;
×
1725
    QUERY_CHECK_CODE(code, lino, end);
×
1726
  }
1727

1728
  //  int64_t st1 = taosGetTimestampUs();
1729
  //  qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
1730
  pBlockList = taosArrayInit(2, POINTER_BYTES);
56,969,555✔
1731
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
56,976,302!
1732

1733
  void* tmp = taosArrayPush(pBlockList, &pResBlock);
56,999,232✔
1734
  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
56,999,232!
1735

1736
  code = createResultData(&type, numOfTables, &output);
56,999,232✔
1737
  if (code != TSDB_CODE_SUCCESS) {
56,945,101!
1738
    terrno = code;
×
1739
    QUERY_CHECK_CODE(code, lino, end);
×
1740
  }
1741

1742
  code = scalarCalculate(pTagCond, pBlockList, &output, pStreamInfo, NULL);
56,945,101✔
1743
  if (code != TSDB_CODE_SUCCESS) {
56,926,096✔
1744
    qError("failed to calculate scalar, reason:%s", tstrerror(code));
4,668!
1745
    terrno = code;
4,668✔
1746
    QUERY_CHECK_CODE(code, lino, end);
4,668!
1747
  }
1748

1749
  code = doSetQualifiedUid(pListInfo, pUidList, pUidTagList, (bool*)output.columnData->pData, addUid);
56,921,428✔
1750
  if (code != TSDB_CODE_SUCCESS) {
56,998,444!
1751
    terrno = code;
×
1752
    QUERY_CHECK_CODE(code, lino, end);
×
1753
  }
1754
  *listAdded = true;
56,998,444✔
1755

1756
end:
61,516,109✔
1757
  if (code != TSDB_CODE_SUCCESS) {
61,512,028✔
1758
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
4,668!
1759
  }
1760
  blockDataDestroy(pResBlock);
61,512,028✔
1761
  taosArrayDestroy(pBlockList);
61,460,330✔
1762
  taosArrayDestroyEx(pUidTagList, freeItem);
61,438,270✔
1763

1764
  colDataDestroy(output.columnData);
61,506,535✔
1765
  taosMemoryFreeClear(output.columnData);
61,506,775!
1766
  return code;
61,494,329✔
1767
}
1768

1769
int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
855,591,044✔
1770
                     STableListInfo* pListInfo, uint8_t* digest, const char* idstr, SStorageAPI* pStorageAPI, void* pStreamInfo) {
1771
  int32_t code = TSDB_CODE_SUCCESS;
855,591,044✔
1772
  int32_t lino = 0;
855,591,044✔
1773
  size_t  numOfTables = 0;
855,591,044✔
1774
  bool    listAdded = false;
855,591,044✔
1775

1776
  pListInfo->idInfo.suid = pScanNode->suid;
855,675,995✔
1777
  pListInfo->idInfo.tableType = pScanNode->tableType;
855,438,000✔
1778

1779
  SArray* pUidList = taosArrayInit(8, sizeof(uint64_t));
855,659,245✔
1780
  QUERY_CHECK_NULL(pUidList, code, lino, _error, terrno);
855,253,095!
1781

1782
  SIdxFltStatus status = SFLT_NOT_INDEX;
855,253,095✔
1783
  if (pScanNode->tableType != TSDB_SUPER_TABLE) {
855,276,168✔
1784
    pListInfo->idInfo.uid = pScanNode->uid;
312,134,000✔
1785
    if (pStorageAPI->metaFn.isTableExisted(pVnode, pScanNode->uid)) {
312,111,606✔
1786
      void* tmp = taosArrayPush(pUidList, &pScanNode->uid);
311,337,126✔
1787
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
311,338,698!
1788
    }
1789
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, false, &listAdded, pStreamInfo);
312,191,620✔
1790
    QUERY_CHECK_CODE(code, lino, _end);
312,168,021!
1791
  } else {
1792
    T_MD5_CTX context = {0};
543,121,662✔
1793

1794
    if (tsTagFilterCache) {
543,392,006!
1795
      // try to retrieve the result from meta cache
UNCOV
1796
      code = genTagFilterDigest(pTagCond, &context);
×
1797
      QUERY_CHECK_CODE(code, lino, _error);
×
1798

1799
      bool acquired = false;
×
1800
      code = pStorageAPI->metaFn.getCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),
×
1801
                                                    pUidList, &acquired);
1802
      QUERY_CHECK_CODE(code, lino, _error);
×
1803

1804
      if (acquired) {
×
1805
        digest[0] = 1;
×
1806
        memcpy(digest + 1, context.digest, tListLen(context.digest));
×
1807
        qDebug("retrieve table uid list from cache, numOfTables:%d", (int32_t)taosArrayGetSize(pUidList));
×
1808
        goto _end;
×
1809
      }
1810
    }
1811

1812
    if (!pTagCond) {  // no tag filter condition exists, let's fetch all tables of this super table
543,448,729✔
1813
      code = pStorageAPI->metaFn.getChildTableList(pVnode, pScanNode->suid, pUidList);
482,997,278✔
1814
      QUERY_CHECK_CODE(code, lino, _error);
482,948,508!
1815
    } else {
1816
      // failed to find the result in the cache, let try to calculate the results
1817
      if (pTagIndexCond) {
60,451,451✔
1818
        void* pIndex = pStorageAPI->metaFn.getInvertIndex(pVnode);
13,359,399✔
1819

1820
        SIndexMetaArg metaArg = {.metaEx = pVnode,
13,347,435✔
1821
                                 .idx = pStorageAPI->metaFn.storeGetIndexInfo(pVnode),
13,350,756✔
1822
                                 .ivtIdx = pIndex,
1823
                                 .suid = pScanNode->uid};
13,356,545✔
1824

1825
        status = SFLT_NOT_INDEX;
13,351,569✔
1826
        code = doFilterTag(pTagIndexCond, &metaArg, pUidList, &status, &pStorageAPI->metaFilter);
13,351,569✔
1827
        if (code != 0 || status == SFLT_NOT_INDEX) {  // temporarily disable it for performance sake
13,349,535✔
1828
          qDebug("failed to get tableIds from index, suid:%" PRIu64, pScanNode->uid);
3,963,714✔
1829
        } else {
1830
          qDebug("succ to get filter result, table num: %d", (int)taosArrayGetSize(pUidList));
9,385,821!
1831
        }
1832
      }
1833
    }
1834

1835
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, tsTagFilterCache, &listAdded, pStreamInfo);
543,404,331✔
1836
    QUERY_CHECK_CODE(code, lino, _end);
543,503,813✔
1837

1838
    // let's add the filter results into meta-cache
1839
    numOfTables = taosArrayGetSize(pUidList);
543,499,145✔
1840

1841
    if (tsTagFilterCache) {
543,386,387!
1842
      size_t size = numOfTables * sizeof(uint64_t) + sizeof(int32_t);
×
1843
      char*  pPayload = taosMemoryMalloc(size);
×
1844
      QUERY_CHECK_NULL(pPayload, code, lino, _end, terrno);
×
1845

1846
      *(int32_t*)pPayload = numOfTables;
×
1847
      if (numOfTables > 0) {
×
1848
        void* tmp = taosArrayGet(pUidList, 0);
×
1849
        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
×
1850
        memcpy(pPayload + sizeof(int32_t), tmp, numOfTables * sizeof(uint64_t));
×
1851
      }
1852

1853
      code = pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),
×
1854
                                                    pPayload, size, 1);
1855
      QUERY_CHECK_CODE(code, lino, _error);
×
1856

1857
      digest[0] = 1;
×
1858
      memcpy(digest + 1, context.digest, tListLen(context.digest));
×
1859
    }
1860
  }
1861

1862
_end:
855,442,548✔
1863
  if (!listAdded) {
855,786,447✔
1864
    numOfTables = taosArrayGetSize(pUidList);
798,521,585✔
1865
    for (int i = 0; i < numOfTables; i++) {
2,147,483,647✔
1866
      void* tmp = taosArrayGet(pUidList, i);
2,147,483,647✔
1867
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
2,147,483,647!
1868
      STableKeyInfo info = {.uid = *(uint64_t*)tmp, .groupId = 0};
2,147,483,647✔
1869

1870
      void* p = taosArrayPush(pListInfo->pTableList, &info);
2,147,483,647✔
1871
      if (p == NULL) {
2,147,483,647!
1872
        taosArrayDestroy(pUidList);
×
1873
        return terrno;
×
1874
      }
1875

1876
      qTrace("tagfilter get uid:%" PRIu64 ", %s", info.uid, idstr);
2,147,483,647!
1877
    }
1878
  }
1879

1880
  qDebug("table list with %d uids built", (int32_t)taosArrayGetSize(pListInfo->pTableList));
855,792,003✔
1881

1882
_error:
855,798,373✔
1883

1884
  taosArrayDestroy(pUidList);
855,847,730✔
1885
  if (code != TSDB_CODE_SUCCESS) {
855,780,199✔
1886
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
4,668!
1887
  }
1888
  return code;
855,766,756✔
1889
}
1890

1891
int32_t qGetTableList(int64_t suid, void* pVnode, void* node, SArray** tableList, void* pTaskInfo) {
65,114✔
1892
  int32_t        code = TSDB_CODE_SUCCESS;
65,114✔
1893
  int32_t        lino = 0;
65,114✔
1894
  SSubplan*      pSubplan = (SSubplan*)node;
65,114✔
1895
  SScanPhysiNode pNode = {0};
65,114✔
1896
  pNode.suid = suid;
65,114✔
1897
  pNode.uid = suid;
65,114✔
1898
  pNode.tableType = TSDB_SUPER_TABLE;
65,114✔
1899

1900
  STableListInfo* pTableListInfo = tableListCreate();
65,114✔
1901
  QUERY_CHECK_NULL(pTableListInfo, code, lino, _end, terrno);
65,114!
1902
  uint8_t digest[17] = {0};
65,114✔
1903
  code = getTableList(pVnode, &pNode, pSubplan ? pSubplan->pTagCond : NULL, pSubplan ? pSubplan->pTagIndexCond : NULL,
65,114✔
1904
                      pTableListInfo, digest, "qGetTableList", &((SExecTaskInfo*)pTaskInfo)->storageAPI, NULL);
1905
  QUERY_CHECK_CODE(code, lino, _end);
65,114!
1906
  *tableList = pTableListInfo->pTableList;
65,114✔
1907
  pTableListInfo->pTableList = NULL;
65,114✔
1908
  tableListDestroy(pTableListInfo);
65,114✔
1909

1910
_end:
65,114✔
1911
  if (code != TSDB_CODE_SUCCESS) {
65,114!
1912
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1913
  }
1914
  return code;
65,114✔
1915
}
1916

1917
size_t getTableTagsBufLen(const SNodeList* pGroups) {
×
1918
  size_t keyLen = 0;
×
1919

1920
  SNode* node;
1921
  FOREACH(node, pGroups) {
×
1922
    SExprNode* pExpr = (SExprNode*)node;
×
1923
    keyLen += pExpr->resType.bytes;
×
1924
  }
1925

1926
  keyLen += sizeof(int8_t) * LIST_LENGTH(pGroups);
×
1927
  return keyLen;
×
1928
}
1929

1930
int32_t getGroupIdFromTagsVal(void* pVnode, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId,
×
1931
                              SStorageAPI* pAPI) {
1932
  SMetaReader mr = {0};
×
1933

1934
  pAPI->metaReaderFn.initReader(&mr, pVnode, META_READER_LOCK, &pAPI->metaFn);
×
1935
  if (pAPI->metaReaderFn.getEntryGetUidCache(&mr, uid) != 0) {  // table not exist
×
1936
    pAPI->metaReaderFn.clearReader(&mr);
×
1937
    return TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
1938
  }
1939

1940
  SNodeList* groupNew = NULL;
×
1941
  int32_t    code = nodesCloneList(pGroupNode, &groupNew);
×
1942
  if (TSDB_CODE_SUCCESS != code) {
×
1943
    pAPI->metaReaderFn.clearReader(&mr);
×
1944
    return code;
×
1945
  }
1946

1947
  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
×
1948
  nodesRewriteExprsPostOrder(groupNew, doTranslateTagExpr, &ctx);
×
1949
  if (TSDB_CODE_SUCCESS != ctx.code) {
×
1950
    nodesDestroyList(groupNew);
×
1951
    pAPI->metaReaderFn.clearReader(&mr);
×
1952
    return code;
×
1953
  }
1954
  char* isNull = (char*)keyBuf;
×
1955
  char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(pGroupNode);
×
1956

1957
  SNode*  pNode;
1958
  int32_t index = 0;
×
1959
  FOREACH(pNode, groupNew) {
×
1960
    SNode*  pNew = NULL;
×
1961
    int32_t code = scalarCalculateConstants(pNode, &pNew);
×
1962
    if (TSDB_CODE_SUCCESS == code) {
×
1963
      REPLACE_NODE(pNew);
×
1964
    } else {
1965
      nodesDestroyList(groupNew);
×
1966
      pAPI->metaReaderFn.clearReader(&mr);
×
1967
      return code;
×
1968
    }
1969

1970
    if (nodeType(pNew) != QUERY_NODE_VALUE) {
×
1971
      nodesDestroyList(groupNew);
×
1972
      pAPI->metaReaderFn.clearReader(&mr);
×
1973
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
1974
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
1975
    }
1976
    SValueNode* pValue = (SValueNode*)pNew;
×
1977

1978
    if (pValue->node.resType.type == TSDB_DATA_TYPE_NULL || pValue->isNull) {
×
1979
      isNull[index++] = 1;
×
1980
      continue;
×
1981
    } else {
1982
      isNull[index++] = 0;
×
1983
      char* data = nodesGetValueFromNode(pValue);
×
1984
      if (pValue->node.resType.type == TSDB_DATA_TYPE_JSON) {
×
1985
        if (tTagIsJson(data)) {
×
1986
          terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
×
1987
          nodesDestroyList(groupNew);
×
1988
          pAPI->metaReaderFn.clearReader(&mr);
×
1989
          return terrno;
×
1990
        }
1991
        int32_t len = getJsonValueLen(data);
×
1992
        memcpy(pStart, data, len);
×
1993
        pStart += len;
×
1994
      } else if (IS_VAR_DATA_TYPE(pValue->node.resType.type)) {
×
1995
        if (IS_STR_DATA_BLOB(pValue->node.resType.type)) {
×
1996
          return TSDB_CODE_BLOB_NOT_SUPPORT_TAG;
×
1997
        }
1998
        memcpy(pStart, data, varDataTLen(data));
×
1999
        pStart += varDataTLen(data);
×
2000
      } else {
2001
        memcpy(pStart, data, pValue->node.resType.bytes);
×
2002
        pStart += pValue->node.resType.bytes;
×
2003
      }
2004
    }
2005
  }
2006

2007
  int32_t len = (int32_t)(pStart - (char*)keyBuf);
×
2008
  *pGroupId = calcGroupId(keyBuf, len);
×
2009

2010
  nodesDestroyList(groupNew);
×
2011
  pAPI->metaReaderFn.clearReader(&mr);
×
2012

2013
  return TSDB_CODE_SUCCESS;
×
2014
}
2015

2016
SArray* makeColumnArrayFromList(SNodeList* pNodeList) {
12,857,354✔
2017
  if (!pNodeList) {
12,857,354!
2018
    return NULL;
×
2019
  }
2020

2021
  size_t  numOfCols = LIST_LENGTH(pNodeList);
12,857,354!
2022
  SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
12,857,354✔
2023
  if (pList == NULL) {
12,854,106!
2024
    return NULL;
×
2025
  }
2026

2027
  for (int32_t i = 0; i < numOfCols; ++i) {
31,598,252✔
2028
    SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i);
18,745,540✔
2029
    if (!pColNode) {
18,747,049!
2030
      taosArrayDestroy(pList);
×
2031
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
2032
      return NULL;
×
2033
    }
2034

2035
    // todo extract method
2036
    SColumn c = {0};
18,747,049✔
2037
    c.slotId = pColNode->slotId;
18,745,540✔
2038
    c.colId = pColNode->colId;
18,746,934✔
2039
    c.type = pColNode->node.resType.type;
18,743,309✔
2040
    c.bytes = pColNode->node.resType.bytes;
18,744,484✔
2041
    c.precision = pColNode->node.resType.precision;
18,745,870✔
2042
    c.scale = pColNode->node.resType.scale;
18,744,023✔
2043

2044
    void* tmp = taosArrayPush(pList, &c);
18,745,089✔
2045
    if (!tmp) {
18,745,089!
2046
      taosArrayDestroy(pList);
×
2047
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2048
      return NULL;
×
2049
    }
2050
  }
2051

2052
  return pList;
12,852,712✔
2053
}
2054

2055
int32_t extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
1,173,446,691✔
2056
                            int32_t type, SColMatchInfo* pMatchInfo) {
2057
  size_t  numOfCols = LIST_LENGTH(pNodeList);
1,173,446,691✔
2058
  int32_t code = TSDB_CODE_SUCCESS;
1,173,472,734✔
2059
  int32_t lino = 0;
1,173,472,734✔
2060

2061
  pMatchInfo->matchType = type;
1,173,472,734✔
2062

2063
  SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchItem));
1,173,452,569✔
2064
  if (pList == NULL) {
1,173,209,265!
2065
    code = terrno;
×
2066
    return code;
×
2067
  }
2068

2069
  for (int32_t i = 0; i < numOfCols; ++i) {
2,147,483,647✔
2070
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);
2,147,483,647✔
2071
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
2,147,483,647!
2072
    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {
2,147,483,647✔
2073
      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
2,147,483,647✔
2074

2075
      SColMatchItem c = {.needOutput = true};
2,147,483,647✔
2076
      c.colId = pColNode->colId;
2,147,483,647✔
2077
      c.srcSlotId = pColNode->slotId;
2,147,483,647✔
2078
      c.dstSlotId = pNode->slotId;
2,147,483,647✔
2079
      c.isPk = pColNode->isPk;
2,147,483,647!
2080
      c.dataType = pColNode->node.resType;
2,147,483,647✔
2081
      void* tmp = taosArrayPush(pList, &c);
2,147,483,647✔
2082
      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
2,147,483,647!
2083
    }
2084
  }
2085

2086
  // set the output flag for each column in SColMatchInfo, according to the
2087
  *numOfOutputCols = 0;
1,173,404,460✔
2088
  int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);
1,173,564,568!
2089
  for (int32_t i = 0; i < num; ++i) {
2,147,483,647✔
2090
    SSlotDescNode* pNode = (SSlotDescNode*)nodesListGetNode(pOutputNodeList->pSlots, i);
2,147,483,647✔
2091
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
2,147,483,647!
2092

2093
    // todo: add reserve flag check
2094
    // it is a column reserved for the arithmetic expression calculation
2095
    if (pNode->slotId >= numOfCols) {
2,147,483,647✔
2096
      (*numOfOutputCols) += 1;
755,276,310✔
2097
      continue;
755,299,580✔
2098
    }
2099

2100
    SColMatchItem* info = NULL;
2,147,483,647✔
2101
    for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {
2,147,483,647✔
2102
      info = taosArrayGet(pList, j);
2,147,483,647✔
2103
      QUERY_CHECK_NULL(info, code, lino, _end, terrno);
2,147,483,647!
2104
      if (info->dstSlotId == pNode->slotId) {
2,147,483,647✔
2105
        break;
2,147,483,647✔
2106
      }
2107
    }
2108

2109
    if (pNode->output) {
7,849,364✔
2110
      (*numOfOutputCols) += 1;
2,147,483,647✔
2111
    } else if (info != NULL) {
46,643,876✔
2112
      // select distinct tbname from stb where tbname='abc';
2113
      info->needOutput = false;
46,569,530✔
2114
    }
2115
  }
2116

2117
  pMatchInfo->pList = pList;
1,173,407,887✔
2118

2119
_end:
1,173,451,487✔
2120
  if (code != TSDB_CODE_SUCCESS) {
1,173,451,487!
2121
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2122
  }
2123
  return code;
1,173,364,825✔
2124
}
2125

2126
static SResSchema createResSchema(int32_t type, int32_t bytes, int32_t slotId, int32_t scale, int32_t precision,
2,147,483,647✔
2127
                                  const char* name) {
2128
  SResSchema s = {0};
2,147,483,647✔
2129
  s.scale = scale;
2,147,483,647✔
2130
  s.type = type;
2,147,483,647✔
2131
  s.bytes = bytes;
2,147,483,647✔
2132
  s.slotId = slotId;
2,147,483,647✔
2133
  s.precision = precision;
2,147,483,647✔
2134
  tstrncpy(s.name, name, tListLen(s.name));
2,147,483,647!
2135

2136
  return s;
2,147,483,647✔
2137
}
2138

2139
static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType, EColumnType colType) {
2,147,483,647✔
2140
  SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn));
2,147,483,647✔
2141
  if (pCol == NULL) {
2,147,483,647!
2142
    return NULL;
×
2143
  }
2144

2145
  pCol->slotId = slotId;
2,147,483,647✔
2146
  pCol->colId = colId;
2,147,483,647✔
2147
  pCol->bytes = pType->bytes;
2,147,483,647✔
2148
  pCol->type = pType->type;
2,147,483,647✔
2149
  pCol->scale = pType->scale;
2,147,483,647✔
2150
  pCol->precision = pType->precision;
2,147,483,647✔
2151
  pCol->dataBlockId = blockId;
2,147,483,647✔
2152
  pCol->colType = colType;
2,147,483,647✔
2153
  return pCol;
2,147,483,647✔
2154
}
2155

2156
int32_t createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) {
2,147,483,647✔
2157
  int32_t code = TSDB_CODE_SUCCESS;
2,147,483,647✔
2158
  int32_t lino = 0;
2,147,483,647✔
2159
  pExp->base.numOfParams = 0;
2,147,483,647✔
2160
  pExp->base.pParam = NULL;
2,147,483,647✔
2161
  pExp->pExpr = taosMemoryCalloc(1, sizeof(tExprNode));
2,147,483,647✔
2162
  QUERY_CHECK_NULL(pExp->pExpr, code, lino, _end, terrno);
2,147,483,647!
2163

2164
  pExp->pExpr->_function.num = 1;
2,147,483,647✔
2165
  pExp->pExpr->_function.functionId = -1;
2,147,483,647✔
2166

2167
  int32_t type = nodeType(pNode);
2,147,483,647✔
2168
  // it is a project query, or group by column
2169
  if (type == QUERY_NODE_COLUMN) {
2,147,483,647✔
2170
    pExp->pExpr->nodeType = QUERY_NODE_COLUMN;
2,147,483,647✔
2171
    SColumnNode* pColNode = (SColumnNode*)pNode;
2,147,483,647✔
2172

2173
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
2,147,483,647✔
2174
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
2,147,483,647!
2175

2176
    pExp->base.numOfParams = 1;
2,147,483,647✔
2177

2178
    SDataType* pType = &pColNode->node.resType;
2,147,483,647✔
2179
    pExp->base.resSchema =
2180
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pColNode->colName);
2,147,483,647✔
2181

2182
    pExp->base.pParam[0].pCol =
2,147,483,647✔
2183
        createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType, pColNode->colType);
2,147,483,647✔
2184
    QUERY_CHECK_NULL(pExp->base.pParam[0].pCol, code, lino, _end, terrno);
2,147,483,647!
2185

2186
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN;
2,147,483,647✔
2187
  } else if (type == QUERY_NODE_VALUE) {
1,886,918,610✔
2188
    pExp->pExpr->nodeType = QUERY_NODE_VALUE;
91,944,450✔
2189
    SValueNode* pValNode = (SValueNode*)pNode;
91,953,627✔
2190

2191
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
91,953,627!
2192
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
91,933,142!
2193

2194
    pExp->base.numOfParams = 1;
91,926,823✔
2195

2196
    SDataType* pType = &pValNode->node.resType;
91,945,958✔
2197
    pExp->base.resSchema =
2198
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pValNode->node.aliasName);
91,940,568✔
2199
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;
91,923,695✔
2200
    code = nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param);
91,938,842✔
2201
    QUERY_CHECK_CODE(code, lino, _end);
91,939,982!
2202
  } else if (type == QUERY_NODE_FUNCTION) {
1,794,974,160✔
2203
    pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;
1,674,796,786✔
2204
    SFunctionNode* pFuncNode = (SFunctionNode*)pNode;
1,674,822,683✔
2205

2206
    SDataType* pType = &pFuncNode->node.resType;
1,674,822,683✔
2207
    pExp->base.resSchema =
2208
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pFuncNode->node.aliasName);
1,674,889,192✔
2209
    tExprNode* pExprNode = pExp->pExpr;
1,674,834,334✔
2210

2211
    pExprNode->_function.functionId = pFuncNode->funcId;
1,674,909,714✔
2212
    pExprNode->_function.pFunctNode = pFuncNode;
1,674,729,101✔
2213
    pExprNode->_function.functionType = pFuncNode->funcType;
1,674,897,147✔
2214

2215
    tstrncpy(pExprNode->_function.functionName, pFuncNode->functionName, tListLen(pExprNode->_function.functionName));
1,674,796,298!
2216

2217
    pExp->base.pParamList = pFuncNode->pParameterList;
1,674,858,464✔
2218
#if 1
2219
    // todo refactor: add the parameter for tbname function
2220
    const char* name = "tbname";
1,674,833,942✔
2221
    int32_t     len = strlen(name);
1,674,833,942!
2222

2223
    if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) &&
1,674,833,942!
2224
        pExprNode->_function.functionName[len] == 0) {
153,868,930✔
2225
      pFuncNode->pParameterList = NULL;
153,836,029✔
2226
      int32_t     code = nodesMakeList(&pFuncNode->pParameterList);
153,833,623✔
2227
      SValueNode* res = NULL;
153,857,123✔
2228
      if (TSDB_CODE_SUCCESS == code) {
153,869,626✔
2229
        code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
153,867,109✔
2230
      }
2231
      QUERY_CHECK_CODE(code, lino, _end);
153,914,262!
2232
      res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
153,914,262✔
2233
      code = nodesListAppend(pFuncNode->pParameterList, (SNode*)res);
153,844,984✔
2234
      if (code != TSDB_CODE_SUCCESS) {
153,875,947!
2235
        nodesDestroyNode((SNode*)res);
×
2236
        res = NULL;
×
2237
      }
2238
      QUERY_CHECK_CODE(code, lino, _end);
153,875,947!
2239
    }
2240
#endif
2241

2242
    int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);
1,675,011,951✔
2243

2244
    pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam));
1,674,928,362✔
2245
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
1,674,586,354!
2246
    pExp->base.numOfParams = numOfParam;
1,674,652,121✔
2247

2248
    for (int32_t j = 0; j < numOfParam && TSDB_CODE_SUCCESS == code; ++j) {
2,147,483,647!
2249
      SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j);
2,147,483,647✔
2250
      QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
2,147,483,647!
2251
      if (p1->type == QUERY_NODE_COLUMN) {
2,147,483,647✔
2252
        SColumnNode* pcn = (SColumnNode*)p1;
1,478,424,385✔
2253

2254
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN;
1,478,424,385✔
2255
        pExp->base.pParam[j].pCol =
2,147,483,647✔
2256
            createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType, pcn->colType);
2,147,483,647✔
2257
        QUERY_CHECK_NULL(pExp->base.pParam[j].pCol, code, lino, _end, terrno);
1,478,390,078!
2258
      } else if (p1->type == QUERY_NODE_VALUE) {
705,897,350✔
2259
        SValueNode* pvn = (SValueNode*)p1;
495,218,845✔
2260
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE;
495,218,845✔
2261
        code = nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param);
495,198,719✔
2262
        QUERY_CHECK_CODE(code, lino, _end);
494,981,096!
2263
      }
2264
    }
2265
    pExp->pExpr->_function.bindExprID = ((SExprNode*)pNode)->bindExprID;
1,674,547,276✔
2266
  } else if (type == QUERY_NODE_OPERATOR) {
120,177,374✔
2267
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
117,059,152✔
2268
    SOperatorNode* pOpNode = (SOperatorNode*)pNode;
117,042,077✔
2269

2270
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
117,042,077✔
2271
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
117,032,586!
2272
    pExp->base.numOfParams = 1;
116,963,308✔
2273

2274
    SDataType* pType = &pOpNode->node.resType;
117,049,302✔
2275
    pExp->base.resSchema =
2276
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pOpNode->node.aliasName);
117,035,048✔
2277
    pExp->pExpr->_optrRoot.pRootNode = pNode;
117,023,585✔
2278
  } else if (type == QUERY_NODE_CASE_WHEN) {
3,118,283✔
2279
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
3,122,724✔
2280
    SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;
3,122,724✔
2281

2282
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
3,122,724!
2283
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
3,122,724!
2284
    pExp->base.numOfParams = 1;
3,122,724✔
2285

2286
    SDataType* pType = &pCaseNode->node.resType;
3,122,724✔
2287
    pExp->base.resSchema =
2288
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCaseNode->node.aliasName);
3,122,724✔
2289
    pExp->pExpr->_optrRoot.pRootNode = pNode;
3,122,724✔
2290
  } else if (type == QUERY_NODE_LOGIC_CONDITION) {
158!
2291
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
12,328✔
2292
    SLogicConditionNode* pCond = (SLogicConditionNode*)pNode;
12,328✔
2293
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
12,328!
2294
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
12,328!
2295
    pExp->base.numOfParams = 1;
12,328✔
2296
    SDataType* pType = &pCond->node.resType;
12,328✔
2297
    pExp->base.resSchema =
2298
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCond->node.aliasName);
12,328✔
2299
    pExp->pExpr->_optrRoot.pRootNode = pNode;
12,328✔
2300
  } else {
UNCOV
2301
    code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
UNCOV
2302
    QUERY_CHECK_CODE(code, lino, _end);
×
2303
  }
2304
  pExp->pExpr->relatedTo = ((SExprNode*)pNode)->relatedTo;
2,147,483,647✔
2305
_end:
2,147,483,647✔
2306
  if (code != TSDB_CODE_SUCCESS) {
2,147,483,647!
2307
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2308
  }
2309
  return code;
2,147,483,647✔
2310
}
2311

2312
int32_t createExprFromTargetNode(SExprInfo* pExp, STargetNode* pTargetNode) {
2,147,483,647✔
2313
  return createExprFromOneNode(pExp, pTargetNode->pExpr, pTargetNode->slotId);
2,147,483,647✔
2314
}
2315

2316
SExprInfo* createExpr(SNodeList* pNodeList, int32_t* numOfExprs) {
×
2317
  *numOfExprs = LIST_LENGTH(pNodeList);
×
2318
  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
×
2319
  if (!pExprs) {
×
2320
    return NULL;
×
2321
  }
2322

2323
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
×
2324
    SExprInfo* pExp = &pExprs[i];
×
2325
    int32_t    code = createExprFromOneNode(pExp, nodesListGetNode(pNodeList, i), i + UD_TAG_COLUMN_INDEX);
×
2326
    if (code != TSDB_CODE_SUCCESS) {
×
2327
      taosMemoryFreeClear(pExprs);
×
2328
      terrno = code;
×
2329
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2330
      return NULL;
×
2331
    }
2332
  }
2333

2334
  return pExprs;
×
2335
}
2336

2337
int32_t createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, SExprInfo** pExprInfo, int32_t* numOfExprs) {
1,760,843,627✔
2338
  QRY_PARAM_CHECK(pExprInfo);
1,760,843,627!
2339

2340
  int32_t code = 0;
1,760,990,765✔
2341
  int32_t numOfFuncs = LIST_LENGTH(pNodeList);
1,760,990,765✔
2342
  int32_t numOfGroupKeys = 0;
1,760,752,574✔
2343
  if (pGroupKeys != NULL) {
1,760,752,574✔
2344
    numOfGroupKeys = LIST_LENGTH(pGroupKeys);
142,678,963!
2345
  }
2346

2347
  *numOfExprs = numOfFuncs + numOfGroupKeys;
1,760,756,210✔
2348
  if (*numOfExprs == 0) {
1,760,924,637✔
2349
    return code;
194,899,997✔
2350
  }
2351

2352
  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
1,565,969,310✔
2353
  if (pExprs == NULL) {
1,565,319,007!
2354
    return terrno;
×
2355
  }
2356

2357
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
2,147,483,647✔
2358
    STargetNode* pTargetNode = NULL;
2,147,483,647✔
2359
    if (i < numOfFuncs) {
2,147,483,647✔
2360
      pTargetNode = (STargetNode*)nodesListGetNode(pNodeList, i);
2,147,483,647✔
2361
    } else {
2362
      pTargetNode = (STargetNode*)nodesListGetNode(pGroupKeys, i - numOfFuncs);
294,446,562✔
2363
    }
2364
    if (!pTargetNode) {
2,147,483,647!
2365
      destroyExprInfo(pExprs, *numOfExprs);
×
2366
      taosMemoryFreeClear(pExprs);
×
2367
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2368
      return terrno;
×
2369
    }
2370

2371
    SExprInfo* pExp = &pExprs[i];
2,147,483,647✔
2372
    code = createExprFromTargetNode(pExp, pTargetNode);
2,147,483,647✔
2373
    if (code != TSDB_CODE_SUCCESS) {
2,147,483,647!
2374
      destroyExprInfo(pExprs, *numOfExprs);
×
2375
      taosMemoryFreeClear(pExprs);
×
2376
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2377
      return code;
×
2378
    }
2379
  }
2380

2381
  *pExprInfo = pExprs;
1,565,896,420✔
2382
  return code;
1,566,017,033✔
2383
}
2384

2385
static void deleteSubsidiareCtx(void* pData) {
×
2386
  SSubsidiaryResInfo* pCtx = (SSubsidiaryResInfo*)pData;
×
2387
  if (pCtx->pCtx) {
×
2388
    taosMemoryFreeClear(pCtx->pCtx);
×
2389
  }
2390
}
×
2391

2392
// set the output buffer for the selectivity + tag query
2393
static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
1,712,344,318✔
2394
  int32_t num = 0;
1,712,344,318✔
2395
  int32_t code = TSDB_CODE_SUCCESS;
1,712,344,318✔
2396
  int32_t lino = 0;
1,712,344,318✔
2397

2398
  SArray* pValCtxArray = NULL;
1,712,344,318✔
2399
  for (int32_t i = numOfOutput - 1; i > 0; --i) {  // select Func is at the end of the list
2,147,483,647✔
2400
    int32_t funcIdx = pCtx[i].pExpr->pExpr->_function.bindExprID;
2,147,483,647✔
2401
    if (funcIdx > 0) {
2,147,483,647✔
2402
      if (pValCtxArray == NULL) {
5,177,767✔
2403
        // the end of the list is the select function of biggest index
2404
        pValCtxArray = taosArrayInit_s(sizeof(SSubsidiaryResInfo*), funcIdx);
3,709,067✔
2405
        if (pValCtxArray == NULL) {
3,710,493!
2406
          return terrno;
×
2407
        }
2408
      }
2409
      if (funcIdx > pValCtxArray->size) {
5,179,193!
2410
        qError("funcIdx:%d is out of range", funcIdx);
×
2411
        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2412
        return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2413
      }
2414
      SSubsidiaryResInfo* pSubsidiary = &pCtx[i].subsidiaries;
5,177,767✔
2415
      pSubsidiary->pCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
5,177,767!
2416
      if (pSubsidiary->pCtx == NULL) {
5,179,193!
2417
        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2418
        return terrno;
×
2419
      }
2420
      pSubsidiary->num = 0;
5,179,193✔
2421
      taosArraySet(pValCtxArray, funcIdx - 1, &pSubsidiary);
5,182,045✔
2422
    }
2423
  }
2424

2425
  SqlFunctionCtx*  p = NULL;
1,711,995,398✔
2426
  SqlFunctionCtx** pValCtx = NULL;
1,711,995,398✔
2427
  if (pValCtxArray == NULL) {
1,711,995,398✔
2428
    pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
1,708,451,901✔
2429
    if (pValCtx == NULL) {
1,708,384,039!
2430
      QUERY_CHECK_CODE(terrno, lino, _end);
×
2431
    }
2432
  }
2433

2434
  for (int32_t i = 0; i < numOfOutput; ++i) {
2,147,483,647✔
2435
    const char* pName = pCtx[i].pExpr->pExpr->_function.functionName;
2,147,483,647✔
2436
    if ((strcmp(pName, "_select_value") == 0)) {
2,147,483,647✔
2437
      if (pValCtxArray == NULL) {
27,953,057✔
2438
        pValCtx[num++] = &pCtx[i];
20,664,735✔
2439
      } else {
2440
        int32_t bindFuncIndex = pCtx[i].pExpr->pExpr->relatedTo;  // start from index 1;
7,288,322✔
2441
        if (bindFuncIndex > 0) {                                  // 0 is default index related to the select function
7,286,896✔
2442
          bindFuncIndex -= 1;
7,110,072✔
2443
        }
2444
        SSubsidiaryResInfo** pSubsidiary = taosArrayGet(pValCtxArray, bindFuncIndex);
7,286,896✔
2445
        if (pSubsidiary == NULL) {
7,288,322!
2446
          QUERY_CHECK_CODE(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, lino, _end);
×
2447
        }
2448
        (*pSubsidiary)->pCtx[(*pSubsidiary)->num] = &pCtx[i];
7,288,322✔
2449
        (*pSubsidiary)->num++;
7,288,322✔
2450
      }
2451
    } else if (fmIsSelectFunc(pCtx[i].functionId)) {
2,147,483,647✔
2452
      if (pValCtxArray == NULL) {
224,855,688✔
2453
        p = &pCtx[i];
218,566,030✔
2454
      }
2455
    }
2456
  }
2457

2458
  if (p != NULL) {
1,712,249,112✔
2459
    p->subsidiaries.pCtx = pValCtx;
103,905,606✔
2460
    p->subsidiaries.num = num;
103,896,170✔
2461
  } else {
2462
    taosMemoryFreeClear(pValCtx);
1,608,343,506✔
2463
  }
2464

2465
_end:
3,730,172✔
2466
  if (code != TSDB_CODE_SUCCESS) {
1,712,036,060!
2467
    taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2468
    taosMemoryFreeClear(pValCtx);
×
2469
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2470
  } else {
2471
    taosArrayDestroy(pValCtxArray);
1,712,036,060✔
2472
  }
2473
  return code;
1,712,178,883✔
2474
}
2475

2476
SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowEntryInfoOffset,
1,712,441,120✔
2477
                                     SFunctionStateStore* pStore) {
2478
  int32_t         code = TSDB_CODE_SUCCESS;
1,712,441,120✔
2479
  int32_t         lino = 0;
1,712,441,120✔
2480
  SqlFunctionCtx* pFuncCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));
1,712,441,120✔
2481
  if (pFuncCtx == NULL) {
1,711,787,613!
2482
    return NULL;
×
2483
  }
2484

2485
  *rowEntryInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t));
1,711,787,613✔
2486
  if (*rowEntryInfoOffset == 0) {
1,712,243,784!
2487
    taosMemoryFreeClear(pFuncCtx);
×
2488
    return NULL;
×
2489
  }
2490

2491
  for (int32_t i = 0; i < numOfOutput; ++i) {
2,147,483,647✔
2492
    SExprInfo* pExpr = &pExprInfo[i];
2,147,483,647✔
2493

2494
    SExprBasicInfo* pFunct = &pExpr->base;
2,147,483,647✔
2495
    SqlFunctionCtx* pCtx = &pFuncCtx[i];
2,147,483,647✔
2496

2497
    pCtx->functionId = -1;
2,147,483,647✔
2498
    pCtx->pExpr = pExpr;
2,147,483,647✔
2499

2500
    if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {
2,147,483,647✔
2501
      SFuncExecEnv env = {0};
1,671,402,547✔
2502
      pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;
1,671,372,175✔
2503
      pCtx->isPseudoFunc = fmIsWindowPseudoColumnFunc(pCtx->functionId) || fmIsPlaceHolderFunc(pCtx->functionId);
1,671,341,094✔
2504
      pCtx->isNotNullFunc = fmIsNotNullOutputFunc(pCtx->functionId);
1,671,352,283✔
2505

2506
      bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);
1,671,293,993✔
2507
      if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId)) {
2,147,483,647✔
2508
        if (!isUdaf) {
1,037,180,294✔
2509
          code = fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
1,037,098,356✔
2510
          QUERY_CHECK_CODE(code, lino, _end);
1,036,873,040!
2511
        } else {
2512
          char* udfName = pExpr->pExpr->_function.pFunctNode->functionName;
81,938✔
2513
          pCtx->udfName = taosStrdup(udfName);
81,938!
2514
          QUERY_CHECK_NULL(pCtx->udfName, code, lino, _end, terrno);
81,938!
2515

2516
          code = fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet);
81,938✔
2517
          QUERY_CHECK_CODE(code, lino, _end);
81,938!
2518
        }
2519
        bool tmp = pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
1,036,954,978✔
2520
        if (!tmp) {
1,036,950,055!
2521
          code = terrno;
×
2522
          QUERY_CHECK_CODE(code, lino, _end);
6,371!
2523
        }
2524
      } else {
2525
        if (fmIsPlaceHolderFunc(pCtx->functionId)) {
633,995,118✔
2526
          code = fmGetStreamPesudoFuncEnv(pCtx->functionId, pExpr->base.pParamList, &env);
36,351,323✔
2527
          QUERY_CHECK_CODE(code, lino, _end);
36,349,986!
2528
        }      
2529
        
2530
        code = fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);
634,126,580✔
2531
        if (code != TSDB_CODE_SUCCESS && isUdaf) {
634,075,376!
2532
          code = TSDB_CODE_SUCCESS;
157,824✔
2533
        }
2534
        QUERY_CHECK_CODE(code, lino, _end);
634,075,376!
2535

2536
        if (pCtx->sfp.getEnv != NULL) {
634,075,376✔
2537
          bool tmp = pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
112,432,016✔
2538
          if (!tmp) {
112,430,665!
2539
            code = terrno;
×
2540
            QUERY_CHECK_CODE(code, lino, _end);
×
2541
          }
2542
        }
2543
      }
2544
      pCtx->resDataInfo.interBufSize = env.calcMemSize;
1,671,075,137✔
2545
    } else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR ||
2,147,483,647✔
2546
               pExpr->pExpr->nodeType == QUERY_NODE_VALUE) {
91,931,367✔
2547
      // for simple column, the result buffer needs to hold at least one element.
2548
      pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes;
2,147,483,647✔
2549
    }
2550

2551
    pCtx->input.numOfInputCols = pFunct->numOfParams;
2,147,483,647✔
2552
    pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
2,147,483,647✔
2553
    QUERY_CHECK_NULL(pCtx->input.pData, code, lino, _end, terrno);
2,147,483,647!
2554
    pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
2,147,483,647✔
2555
    QUERY_CHECK_NULL(pCtx->input.pColumnDataAgg, code, lino, _end, terrno);
2,147,483,647!
2556

2557
    pCtx->pTsOutput = NULL;
2,147,483,647✔
2558
    pCtx->resDataInfo.bytes = pFunct->resSchema.bytes;
2,147,483,647✔
2559
    pCtx->resDataInfo.type = pFunct->resSchema.type;
2,147,483,647✔
2560
    pCtx->order = TSDB_ORDER_ASC;
2,147,483,647✔
2561
    pCtx->start.key = INT64_MIN;
2,147,483,647✔
2562
    pCtx->end.key = INT64_MIN;
2,147,483,647✔
2563
    pCtx->numOfParams = pExpr->base.numOfParams;
2,147,483,647✔
2564
    pCtx->param = pFunct->pParam;
2,147,483,647✔
2565
    pCtx->saveHandle.currentPage = -1;
2,147,483,647✔
2566
    pCtx->pStore = pStore;
2,147,483,647✔
2567
    pCtx->hasWindowOrGroup = false;
2,147,483,647✔
2568
    pCtx->needCleanup = false;
2,147,483,647✔
2569
  }
2570

2571
  for (int32_t i = 1; i < numOfOutput; ++i) {
2,147,483,647✔
2572
    (*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) +
2,147,483,647✔
2573
                                         pFuncCtx[i - 1].resDataInfo.interBufSize);
2,147,483,647✔
2574
  }
2575

2576
  code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);
1,712,314,894✔
2577
  QUERY_CHECK_CODE(code, lino, _end);
1,712,126,778!
2578

2579
_end:
1,712,126,778✔
2580
  if (code != TSDB_CODE_SUCCESS) {
1,712,027,314!
2581
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2582
    for (int32_t i = 0; i < numOfOutput; ++i) {
×
2583
      taosMemoryFree(pFuncCtx[i].input.pData);
×
2584
      taosMemoryFree(pFuncCtx[i].input.pColumnDataAgg);
×
2585
    }
2586
    taosMemoryFreeClear(*rowEntryInfoOffset);
×
2587
    taosMemoryFreeClear(pFuncCtx);
×
2588

2589
    terrno = code;
×
2590
    return NULL;
×
2591
  }
2592
  return pFuncCtx;
1,712,027,314✔
2593
}
2594

2595
// NOTE: sources columns are more than the destination SSDatablock columns.
2596
// doFilter in table scan needs every column even its output is false
2597
int32_t relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols, bool outputEveryColumn) {
44,698,092✔
2598
  int32_t code = TSDB_CODE_SUCCESS;
44,698,092✔
2599
  size_t  numOfSrcCols = taosArrayGetSize(pCols);
44,698,092✔
2600

2601
  int32_t i = 0, j = 0;
44,698,775✔
2602
  while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) {
295,852,206✔
2603
    SColumnInfoData* p = taosArrayGet(pCols, i);
251,152,675✔
2604
    if (!p) {
251,153,986!
2605
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2606
      return terrno;
×
2607
    }
2608
    SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);
251,153,986✔
2609
    if (!pmInfo) {
251,150,576!
2610
      return terrno;
×
2611
    }
2612

2613
    if (p->info.colId == pmInfo->colId) {
251,150,576✔
2614
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->dstSlotId);
223,924,592✔
2615
      if (!pDst) {
223,925,275!
2616
        return terrno;
×
2617
      }
2618
      code = colDataAssign(pDst, p, pBlock->info.rows, &pBlock->info);
223,925,275✔
2619
      if (code != TSDB_CODE_SUCCESS) {
223,924,441!
2620
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2621
        return code;
×
2622
      }
2623
      i++;
223,924,441✔
2624
      j++;
223,924,441✔
2625
    } else if (p->info.colId < pmInfo->colId) {
27,228,990!
2626
      i++;
27,228,990✔
2627
    } else {
2628
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
2629
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2630
    }
2631
  }
2632
  return code;
44,700,214✔
2633
}
2634

2635
SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
717,875,305✔
2636
  SInterval interval = {
1,435,084,302✔
2637
      .interval = pTableScanNode->interval,
717,607,792✔
2638
      .sliding = pTableScanNode->sliding,
717,551,737✔
2639
      .intervalUnit = pTableScanNode->intervalUnit,
717,972,737✔
2640
      .slidingUnit = pTableScanNode->slidingUnit,
717,646,519✔
2641
      .offset = pTableScanNode->offset,
717,948,046✔
2642
      .precision = pTableScanNode->scan.node.pOutputDataBlockDesc->precision,
717,897,049✔
2643
      .timeRange = pTableScanNode->scanRange,
2644
  };
2645
  calcIntervalAutoOffset(&interval);
717,542,199✔
2646

2647
  return interval;
717,500,990✔
2648
}
2649

2650
SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {
309,741,859✔
2651
  SColumn c = {0};
309,741,859✔
2652

2653
  c.slotId = pColNode->slotId;
309,741,859✔
2654
  c.colId = pColNode->colId;
309,739,351✔
2655
  c.type = pColNode->node.resType.type;
309,746,183✔
2656
  c.bytes = pColNode->node.resType.bytes;
309,733,790✔
2657
  c.scale = pColNode->node.resType.scale;
309,742,034✔
2658
  c.precision = pColNode->node.resType.precision;
309,749,167✔
2659
  return c;
309,722,000✔
2660
}
2661

2662

2663
/**
2664
 * @brief Determine the actual time range for reading data based on the RANGE clause and the WHERE conditions.
2665
 * @param[in] cond The range specified by WHERE condition.
2666
 * @param[in] range The range specified by RANGE clause.
2667
 * @param[out] twindow The range to be read in DESC order, and only one record is needed.
2668
 * @param[out] extTwindow The external range to read for only one record, which is used for FILL clause.
2669
 * @note `cond` and `twindow` may be the same address.
2670
 */
2671
static int32_t getQueryExtWindow(const STimeWindow* cond, const STimeWindow* range, STimeWindow* twindow,
12,905,381✔
2672
                                 STimeWindow* extTwindows) {
2673
  int32_t     code = TSDB_CODE_SUCCESS;
12,905,381✔
2674
  int32_t     lino = 0;
12,905,381✔
2675
  STimeWindow tempWindow;
2676

2677
  if (cond->skey > cond->ekey || range->skey > range->ekey) {
12,905,381!
2678
    *twindow = extTwindows[0] = extTwindows[1] = TSWINDOW_DESC_INITIALIZER;
13,670✔
2679
    return code;
13,670✔
2680
  }
2681

2682
  if (range->ekey < cond->skey) {
12,890,996✔
2683
    extTwindows[1] = *cond;
2,208,877✔
2684
    *twindow = extTwindows[0] = TSWINDOW_DESC_INITIALIZER;
2,208,877✔
2685
    return code;
2,208,877✔
2686
  }
2687

2688
  if (cond->ekey < range->skey) {
10,682,119✔
2689
    extTwindows[0] = *cond;
1,515,497✔
2690
    *twindow = extTwindows[1] = TSWINDOW_DESC_INITIALIZER;
1,515,497✔
2691
    return code;
1,515,497✔
2692
  }
2693

2694
  // Only scan data in the time range intersecion.
2695
  extTwindows[0] = extTwindows[1] = *cond;
9,166,622✔
2696
  twindow->skey = TMAX(cond->skey, range->skey);
9,166,622✔
2697
  twindow->ekey = TMIN(cond->ekey, range->ekey);
9,166,622✔
2698
  extTwindows[0].ekey = twindow->skey - 1;
9,167,337✔
2699
  extTwindows[1].skey = twindow->ekey + 1;
9,166,622✔
2700

2701
  return code;
9,166,622✔
2702
}
2703

2704
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode,
873,485,274✔
2705
                               const SReadHandle* readHandle, bool applyExtWin) {
2706
  int32_t code = 0;                             
873,485,274✔
2707
  pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
873,485,274✔
2708
  pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);
873,538,756!
2709

2710
  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
873,557,858✔
2711
  if (!pCond->colList) {
873,374,644!
2712
    return terrno;
×
2713
  }
2714
  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
873,241,033✔
2715
  if (pCond->pSlotList == NULL) {
873,377,921!
2716
    taosMemoryFreeClear(pCond->colList);
×
2717
    return terrno;
×
2718
  }
2719

2720
  // TODO: get it from stable scan node
2721
  pCond->twindows = pTableScanNode->scanRange;
873,024,992✔
2722
  pCond->suid = pTableScanNode->scan.suid;
873,496,702✔
2723
  pCond->type = TIMEWINDOW_RANGE_CONTAINED;
873,153,197✔
2724
  pCond->startVersion = -1;
873,396,226✔
2725
  pCond->endVersion = -1;
873,485,964✔
2726
  pCond->skipRollup = readHandle->skipRollup;
873,236,247!
2727
  if (readHandle->winRangeValid) {
873,366,135✔
2728
    pCond->twindows = readHandle->winRange;
1,209,165✔
2729
  }
2730
  // allowed read stt file optimization mode
2731
  pCond->notLoadData = (pTableScanNode->dataRequired == FUNC_DATA_REQUIRED_NOT_LOAD) &&
1,747,033,394✔
2732
                       (pTableScanNode->scan.node.pConditions == NULL) && (pTableScanNode->interval == 0);
873,496,411✔
2733

2734
  int32_t j = 0;
873,115,278✔
2735
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
2,147,483,647✔
2736
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pTableScanNode->scan.pScanCols, i);
2,147,483,647✔
2737
    if (!pNode) {
2,147,483,647!
2738
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2739
      return terrno;
×
2740
    }
2741
    SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
2,147,483,647✔
2742
    if (pColNode->colType == COLUMN_TYPE_TAG) {
2,147,483,647!
2743
      continue;
×
2744
    }
2745

2746
    pCond->colList[j].type = pColNode->node.resType.type;
2,147,483,647✔
2747
    pCond->colList[j].bytes = pColNode->node.resType.bytes;
2,147,483,647✔
2748
    pCond->colList[j].colId = pColNode->colId;
2,147,483,647✔
2749
    pCond->colList[j].pk = pColNode->isPk;
2,147,483,647!
2750

2751
    pCond->pSlotList[j] = pNode->slotId;
2,147,483,647✔
2752
    j += 1;
2,147,483,647✔
2753
  }
2754

2755
  pCond->numOfCols = j;
873,434,347✔
2756

2757
  if (applyExtWin) {
873,609,138✔
2758
    if (NULL != pTableScanNode->pExtScanRange) {
719,192,338✔
2759
      pCond->type = TIMEWINDOW_RANGE_EXTERNAL;
12,906,096✔
2760
      code = getQueryExtWindow(&pCond->twindows, pTableScanNode->pExtScanRange, &pCond->twindows, pCond->extTwindows);
12,905,381✔
2761
    } else if (readHandle->extWinRangeValid) {
705,912,591!
2762
      pCond->type = TIMEWINDOW_RANGE_EXTERNAL;
×
2763
      code = getQueryExtWindow(&pCond->twindows, &readHandle->extWinRange, &pCond->twindows, pCond->extTwindows);
×
2764
    }
2765
  }
2766
  
2767
  return code;
873,371,253✔
2768
}
2769

2770
int32_t initQueryTableDataCondWithColArray(SQueryTableDataCond* pCond, SQueryTableDataCond* pOrgCond,
65,872,227✔
2771
                                           const SReadHandle* readHandle, SArray* colArray) {
2772
  int32_t code = TSDB_CODE_SUCCESS;
65,872,227✔
2773
  int32_t lino = 0;
65,872,227✔
2774

2775
  pCond->order = TSDB_ORDER_ASC;
65,872,227✔
2776
  pCond->numOfCols = (int32_t)taosArrayGetSize(colArray);
65,872,227✔
2777

2778
  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
65,872,227!
2779
  QUERY_CHECK_NULL(pCond->colList, code, lino, _return, terrno);
65,872,227!
2780

2781
  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
65,872,227!
2782
  QUERY_CHECK_NULL(pCond->pSlotList, code, lino, _return, terrno);
65,872,227!
2783

2784
  pCond->twindows = pOrgCond->twindows;
65,872,227✔
2785
  pCond->type = pOrgCond->type;
65,872,227✔
2786
  pCond->startVersion = -1;
65,872,227✔
2787
  pCond->endVersion = -1;
65,872,227✔
2788
  pCond->skipRollup = true;
65,872,227✔
2789
  pCond->notLoadData = false;
65,872,227✔
2790

2791
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
235,110,513✔
2792
    SColIdPair* pColPair = taosArrayGet(colArray, i);
169,238,286✔
2793
    QUERY_CHECK_NULL(pColPair, code, lino, _return, terrno);
169,238,286!
2794

2795
    bool find = false;
169,238,286✔
2796
    for (int32_t j = 0; j < pOrgCond->numOfCols; ++j) {
839,689,802!
2797
      if (pOrgCond->colList[j].colId == pColPair->vtbColId) {
839,689,802✔
2798
        pCond->colList[i].type = pOrgCond->colList[j].type;
169,238,286✔
2799
        pCond->colList[i].bytes = pOrgCond->colList[j].bytes;
169,238,286✔
2800
        pCond->colList[i].colId = pColPair->orgColId;
169,238,286✔
2801
        pCond->colList[i].pk = pOrgCond->colList[j].pk;
169,238,286✔
2802
        pCond->pSlotList[i] = i;
169,238,286✔
2803
        find = true;
169,238,286✔
2804
        break;
169,238,286✔
2805
      }
2806
    }
2807
    QUERY_CHECK_CONDITION(find, code, lino, _return, TSDB_CODE_NOT_FOUND);
169,238,286!
2808
  }
2809

2810
  return code;
65,872,227✔
2811
_return:
×
2812
  qError("%s failed at line %d since %s", __func__, lino, tstrerror(terrno));
×
2813
  taosMemoryFreeClear(pCond->colList);
×
2814
  taosMemoryFreeClear(pCond->pSlotList);
×
2815
  return code;
×
2816
}
2817

2818
void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {
1,891,169,605✔
2819
  taosMemoryFreeClear(pCond->colList);
1,891,169,605✔
2820
  taosMemoryFreeClear(pCond->pSlotList);
1,891,176,277✔
2821
}
1,891,157,425✔
2822

2823
int32_t convertFillType(int32_t mode) {
16,534,218✔
2824
  int32_t type = TSDB_FILL_NONE;
16,534,218✔
2825
  switch (mode) {
16,534,218!
2826
    case FILL_MODE_PREV:
707,071✔
2827
      type = TSDB_FILL_PREV;
707,071✔
2828
      break;
707,071✔
2829
    case FILL_MODE_NONE:
×
2830
      type = TSDB_FILL_NONE;
×
2831
      break;
×
2832
    case FILL_MODE_NULL:
872,031✔
2833
      type = TSDB_FILL_NULL;
872,031✔
2834
      break;
872,031✔
2835
    case FILL_MODE_NULL_F:
334,406✔
2836
      type = TSDB_FILL_NULL_F;
334,406✔
2837
      break;
334,406✔
2838
    case FILL_MODE_NEXT:
717,788✔
2839
      type = TSDB_FILL_NEXT;
717,788✔
2840
      break;
717,788✔
2841
    case FILL_MODE_VALUE:
502,877✔
2842
      type = TSDB_FILL_SET_VALUE;
502,877✔
2843
      break;
502,877✔
2844
    case FILL_MODE_VALUE_F:
21,346✔
2845
      type = TSDB_FILL_SET_VALUE_F;
21,346✔
2846
      break;
21,346✔
2847
    case FILL_MODE_LINEAR:
1,269,669✔
2848
      type = TSDB_FILL_LINEAR;
1,269,669✔
2849
      break;
1,269,669✔
2850
    case FILL_MODE_NEAR:
12,109,030✔
2851
      type = TSDB_FILL_NEAR;
12,109,030✔
2852
      break;
12,109,030✔
2853
    default:
×
2854
      type = TSDB_FILL_NONE;
×
2855
  }
2856

2857
  return type;
16,534,218✔
2858
}
2859

2860
void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery) {
14,919,209✔
2861
  if (ascQuery) {
14,919,209✔
2862
    *w = getAlignQueryTimeWindow(pInterval, ts);
11,747,372✔
2863
  } else {
2864
    // the start position of the first time window in the endpoint that spreads beyond the queried last timestamp
2865
    *w = getAlignQueryTimeWindow(pInterval, ts);
3,171,837✔
2866

2867
    int64_t key = w->skey;
3,177,405✔
2868
    while (key < ts) {  // moving towards end
3,243,395✔
2869
      key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);
1,154,255✔
2870
      if (key > ts) {
1,154,255✔
2871
        break;
1,088,265✔
2872
      }
2873

2874
      w->skey = key;
65,990✔
2875
    }
2876
    w->ekey = taosTimeAdd(w->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
3,177,405✔
2877
  }
2878
}
14,926,152✔
2879

2880
static STimeWindow doCalculateTimeWindow(int64_t ts, SInterval* pInterval) {
678,363,433✔
2881
  STimeWindow w = {0};
678,363,433✔
2882

2883
  w.skey = taosTimeTruncate(ts, pInterval);
678,363,433✔
2884
  w.ekey = taosTimeGetIntervalEnd(w.skey, pInterval);
678,366,512✔
2885
  return w;
678,442,014✔
2886
}
2887

2888
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {
6,949,228✔
2889
  STimeWindow win = *pWindow;
6,949,228✔
2890
  STimeWindow save = win;
6,949,228✔
2891
  while (win.skey <= ts && win.ekey >= ts) {
36,698,981✔
2892
    save = win;
29,749,753✔
2893
    // get previous time window
2894
    getNextTimeWindow(pInterval, &win, order == TSDB_ORDER_ASC ? TSDB_ORDER_DESC : TSDB_ORDER_ASC);
29,749,753✔
2895
  }
2896

2897
  return save;
6,949,228✔
2898
}
2899

2900
// get the correct time window according to the handled timestamp
2901
// todo refactor
2902
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
896,914,820✔
2903
                                int32_t order) {
2904
  STimeWindow w = {0};
896,914,820✔
2905
  if (pResultRowInfo->cur.pageId == -1) {  // the first window, from the previous stored value
896,991,980✔
2906
    getInitialStartTimeWindow(pInterval, ts, &w, (order == TSDB_ORDER_ASC));
14,918,065✔
2907
    return w;
14,919,209✔
2908
  }
2909

2910
  SResultRow* pRow = getResultRowByPos(pBuf, &pResultRowInfo->cur, false);
882,107,246✔
2911
  if (pRow) {
882,105,799✔
2912
    TAOS_SET_OBJ_ALIGNED(&w, pRow->win);
882,143,667✔
2913
  }
2914

2915
  // in case of typical time window, we can calculate time window directly.
2916
  if (w.skey > ts || w.ekey < ts) {
882,144,574✔
2917
    w = doCalculateTimeWindow(ts, pInterval);
678,389,662✔
2918
  }
2919

2920
  if (pInterval->interval != pInterval->sliding) {
882,201,206✔
2921
    // it is an sliding window query, in which sliding value is not equalled to
2922
    // interval value, and we need to find the first qualified time window.
2923
    w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
6,949,228✔
2924
  }
2925

2926
  return w;
882,048,387✔
2927
}
2928

2929
TSKEY getNextTimeWindowStart(const SInterval* pInterval, TSKEY start, int32_t order) {
2,147,483,647✔
2930
  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
2,147,483,647✔
2931
  TSKEY   nextStart = taosTimeAdd(start, -1 * pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
2,147,483,647✔
2932
  nextStart = taosTimeAdd(nextStart, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision, NULL);
2,147,483,647✔
2933
  nextStart = taosTimeAdd(nextStart, pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
2,147,483,647✔
2934
  return nextStart;
2,147,483,647✔
2935
}
2936

2937
void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t order) {
2,147,483,647✔
2938
  tw->skey = getNextTimeWindowStart(pInterval, tw->skey, order);
2,147,483,647✔
2939
  tw->ekey = taosTimeAdd(tw->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
2,147,483,647✔
2940
}
2,147,483,647✔
2941

2942
bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {
2,147,483,647✔
2943
  return (pLimitInfo->limit.limit != -1 || pLimitInfo->limit.offset != -1 || pLimitInfo->slimit.limit != -1 ||
2,147,483,647!
2944
          pLimitInfo->slimit.offset != -1);
2,147,483,647✔
2945
}
2946

2947
bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo) {
×
2948
  return (pLimitInfo->slimit.limit != -1 || pLimitInfo->slimit.offset != -1);
×
2949
}
2950

2951
void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo) {
2,147,483,647✔
2952
  SLimit limit = {.limit = getLimit(pLimit), .offset = getOffset(pLimit)};
2,147,483,647✔
2953
  SLimit slimit = {.limit = getLimit(pSLimit), .offset = getOffset(pSLimit)};
2,147,483,647✔
2954

2955
  pLimitInfo->limit = limit;
2,147,483,647✔
2956
  pLimitInfo->slimit = slimit;
2,147,483,647✔
2957
  pLimitInfo->remainOffset = limit.offset;
2,147,483,647✔
2958
  pLimitInfo->remainGroupOffset = slimit.offset;
2,147,483,647✔
2959
  pLimitInfo->numOfOutputRows = 0;
2,147,483,647✔
2960
  pLimitInfo->numOfOutputGroups = 0;
2,147,483,647✔
2961
  pLimitInfo->currentGroupId = 0;
2,147,483,647✔
2962
}
2,147,483,647✔
2963

2964
void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo) {
513,417,104✔
2965
  pLimitInfo->numOfOutputRows = 0;
513,417,104✔
2966
  pLimitInfo->remainOffset = pLimitInfo->limit.offset;
513,436,874✔
2967
}
513,425,818✔
2968

2969
int32_t tableListGetSize(const STableListInfo* pTableList, int32_t* pRes) {
2,147,483,647✔
2970
  if (taosArrayGetSize(pTableList->pTableList) != taosHashGetSize(pTableList->map)) {
2,147,483,647!
2971
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
2972
    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2973
  }
2974
  (*pRes) = taosArrayGetSize(pTableList->pTableList);
2,147,483,647✔
2975
  return TSDB_CODE_SUCCESS;
2,147,483,647✔
2976
}
2977

2978
uint64_t tableListGetSuid(const STableListInfo* pTableList) { return pTableList->idInfo.suid; }
16,089,518✔
2979

2980
STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index) {
965,148,722✔
2981
  if (taosArrayGetSize(pTableList->pTableList) == 0) {
965,148,722✔
2982
    return NULL;
13,206✔
2983
  }
2984

2985
  return taosArrayGet(pTableList->pTableList, index);
964,980,776✔
2986
}
2987

2988
int32_t tableListFind(const STableListInfo* pTableList, uint64_t uid, int32_t startIndex) {
131,004✔
2989
  int32_t numOfTables = taosArrayGetSize(pTableList->pTableList);
131,004✔
2990
  if (startIndex >= numOfTables) {
131,004!
2991
    return -1;
×
2992
  }
2993

2994
  for (int32_t i = startIndex; i < numOfTables; ++i) {
1,229,123!
2995
    STableKeyInfo* p = taosArrayGet(pTableList->pTableList, i);
1,229,123✔
2996
    if (!p) {
1,229,123!
2997
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2998
      return -1;
×
2999
    }
3000
    if (p->uid == uid) {
1,229,123✔
3001
      return i;
131,004✔
3002
    }
3003
  }
3004
  return -1;
×
3005
}
3006

3007
void tableListGetSourceTableInfo(const STableListInfo* pTableList, uint64_t* psuid, uint64_t* uid, int32_t* type) {
1,063,262✔
3008
  *psuid = pTableList->idInfo.suid;
1,063,262✔
3009
  *uid = pTableList->idInfo.uid;
1,063,262✔
3010
  *type = pTableList->idInfo.tableType;
1,063,262✔
3011
}
1,063,262✔
3012

3013
uint64_t tableListGetTableGroupId(const STableListInfo* pTableList, uint64_t tableUid) {
2,147,483,647✔
3014
  int32_t* slot = taosHashGet(pTableList->map, &tableUid, sizeof(tableUid));
2,147,483,647✔
3015
  if (slot == NULL) {
2,147,483,647✔
3016
    qDebug("table:%" PRIu64 " not found in table list", tableUid);
9,682,066✔
3017
    return -1;
9,683,409✔
3018
  }
3019

3020
  STableKeyInfo* pKeyInfo = taosArrayGet(pTableList->pTableList, *slot);
2,147,483,647✔
3021
  return pKeyInfo->groupId;
2,147,483,647✔
3022
}
3023

3024
// TODO handle the group offset info, fix it, the rule of group output will be broken by this function
3025
int32_t tableListAddTableInfo(STableListInfo* pTableList, uint64_t uid, uint64_t gid) {
13,447,363✔
3026
  int32_t code = TSDB_CODE_SUCCESS;
13,447,363✔
3027
  int32_t lino = 0;
13,447,363✔
3028
  if (pTableList->map == NULL) {
13,447,363!
3029
    pTableList->map = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
×
3030
    QUERY_CHECK_NULL(pTableList->map, code, lino, _end, terrno);
×
3031
  }
3032

3033
  STableKeyInfo keyInfo = {.uid = uid, .groupId = gid};
13,447,363✔
3034
  void*         p = taosHashGet(pTableList->map, &uid, sizeof(uid));
13,443,517✔
3035
  if (p != NULL) {
13,441,385✔
3036
    qInfo("table:%" PRId64 " already in tableIdList, ignore it", uid);
2,585!
3037
    goto _end;
2,585✔
3038
  }
3039

3040
  void* tmp = taosArrayPush(pTableList->pTableList, &keyInfo);
13,438,800✔
3041
  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
13,439,218!
3042

3043
  int32_t slot = (int32_t)taosArrayGetSize(pTableList->pTableList) - 1;
13,439,218✔
3044
  code = taosHashPut(pTableList->map, &uid, sizeof(uid), &slot, sizeof(slot));
13,441,371✔
3045
  if (code != TSDB_CODE_SUCCESS) {
13,444,360!
3046
    // we have checked the existence of uid in hash map above
3047
    QUERY_CHECK_CONDITION((code != TSDB_CODE_DUP_KEY), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
×
3048
    taosArrayPopTailBatch(pTableList->pTableList, 1);  // let's pop the last element in the array list
×
3049
  }
3050

3051
_end:
13,446,945✔
3052
  if (code != TSDB_CODE_SUCCESS) {
13,444,792!
3053
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
3054
  } else {
3055
    qDebug("uid:%" PRIu64 ", groupId:%" PRIu64 " added into table list, slot:%d, total:%d", uid, gid, slot, slot + 1);
13,444,792✔
3056
  }
3057

3058
  return code;
13,449,077✔
3059
}
3060

3061
int32_t tableListGetGroupList(const STableListInfo* pTableList, int32_t ordinalGroupIndex, STableKeyInfo** pKeyInfo,
826,144,301✔
3062
                              int32_t* size) {
3063
  int32_t totalGroups = tableListGetOutputGroups(pTableList);
826,144,301✔
3064
  int32_t numOfTables = 0;
826,155,146✔
3065
  int32_t code = tableListGetSize(pTableList, &numOfTables);
826,238,348✔
3066
  if (code != TSDB_CODE_SUCCESS) {
826,122,708!
3067
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
3068
    return code;
×
3069
  }
3070

3071
  if (ordinalGroupIndex < 0 || ordinalGroupIndex >= totalGroups) {
826,122,708✔
3072
    return TSDB_CODE_INVALID_PARA;
×
3073
  }
3074

3075
  // here handle two special cases:
3076
  // 1. only one group exists, and 2. one table exists for each group.
3077
  if (totalGroups == 1) {
826,122,708✔
3078
    *size = numOfTables;
819,824,574✔
3079
    *pKeyInfo = (*size == 0) ? NULL : taosArrayGet(pTableList->pTableList, 0);
819,748,506✔
3080
    return TSDB_CODE_SUCCESS;
819,832,425✔
3081
  } else if (totalGroups == numOfTables) {
6,298,295✔
3082
    *size = 1;
5,664,982✔
3083
    *pKeyInfo = taosArrayGet(pTableList->pTableList, ordinalGroupIndex);
5,664,982✔
3084
    return TSDB_CODE_SUCCESS;
5,665,087✔
3085
  }
3086

3087
  int32_t offset = pTableList->groupOffset[ordinalGroupIndex];
633,353✔
3088
  if (ordinalGroupIndex < totalGroups - 1) {
674,823✔
3089
    *size = pTableList->groupOffset[ordinalGroupIndex + 1] - offset;
531,825✔
3090
  } else {
3091
    *size = numOfTables - offset;
142,998✔
3092
  }
3093

3094
  *pKeyInfo = taosArrayGet(pTableList->pTableList, offset);
674,823✔
3095
  return TSDB_CODE_SUCCESS;
674,823✔
3096
}
3097

3098
int32_t tableListGetOutputGroups(const STableListInfo* pTableList) { return pTableList->numOfOuputGroups; }
2,147,483,647✔
3099

3100
bool oneTableForEachGroup(const STableListInfo* pTableList) { return pTableList->oneTableForEachGroup; }
3,258,860!
3101

3102
STableListInfo* tableListCreate() {
929,138,681✔
3103
  STableListInfo* pListInfo = taosMemoryCalloc(1, sizeof(STableListInfo));
929,138,681✔
3104
  if (pListInfo == NULL) {
928,836,986!
3105
    return NULL;
×
3106
  }
3107

3108
  pListInfo->remainGroups = NULL;
928,836,986✔
3109
  pListInfo->pTableList = taosArrayInit(4, sizeof(STableKeyInfo));
928,872,317✔
3110
  if (pListInfo->pTableList == NULL) {
928,948,420!
3111
    goto _error;
×
3112
  }
3113

3114
  pListInfo->map = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
929,031,710✔
3115
  if (pListInfo->map == NULL) {
929,336,836!
3116
    goto _error;
×
3117
  }
3118

3119
  pListInfo->numOfOuputGroups = 1;
929,332,887✔
3120
  return pListInfo;
929,338,891✔
3121

3122
_error:
×
3123
  tableListDestroy(pListInfo);
×
3124
  return NULL;
×
3125
}
3126

3127
void tableListDestroy(STableListInfo* pTableListInfo) {
930,017,147✔
3128
  if (pTableListInfo == NULL) {
930,017,147✔
3129
    return;
805,439✔
3130
  }
3131

3132
  taosArrayDestroy(pTableListInfo->pTableList);
929,211,708✔
3133
  taosMemoryFreeClear(pTableListInfo->groupOffset);
929,148,027✔
3134

3135
  taosHashCleanup(pTableListInfo->map);
929,169,639✔
3136
  taosHashCleanup(pTableListInfo->remainGroups);
929,285,562✔
3137
  pTableListInfo->pTableList = NULL;
929,270,502✔
3138
  pTableListInfo->map = NULL;
929,299,962✔
3139
  taosMemoryFree(pTableListInfo);
929,298,818✔
3140
}
3141

3142
void tableListClear(STableListInfo* pTableListInfo) {
2,222,376✔
3143
  if (pTableListInfo == NULL) {
2,222,376!
3144
    return;
×
3145
  }
3146

3147
  taosArrayClear(pTableListInfo->pTableList);
2,222,376✔
3148
  taosHashClear(pTableListInfo->map);
2,222,304✔
3149
  taosHashClear(pTableListInfo->remainGroups);
2,226,661✔
3150
  taosMemoryFree(pTableListInfo->groupOffset);
2,226,661!
3151
  pTableListInfo->numOfOuputGroups = 1;
2,226,661✔
3152
  pTableListInfo->oneTableForEachGroup = false;
2,226,661✔
3153
}
3154

3155
static int32_t orderbyGroupIdComparFn(const void* p1, const void* p2) {
2,147,483,647✔
3156
  STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;
2,147,483,647✔
3157
  STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;
2,147,483,647✔
3158

3159
  if (pInfo1->groupId == pInfo2->groupId) {
2,147,483,647✔
3160
    return 0;
1,968,830,270✔
3161
  } else {
3162
    return pInfo1->groupId < pInfo2->groupId ? -1 : 1;
314,107,346✔
3163
  }
3164
}
3165

3166
static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
169,162,689✔
3167
  taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
169,162,689✔
3168
  int32_t size = taosArrayGetSize(pTableListInfo->pTableList);
169,210,502✔
3169

3170
  SArray* pList = taosArrayInit(4, sizeof(int32_t));
169,213,979✔
3171
  if (!pList) {
169,186,136!
3172
    return terrno;
×
3173
  }
3174

3175
  STableKeyInfo* pInfo = taosArrayGet(pTableListInfo->pTableList, 0);
169,186,136✔
3176
  if (!pInfo) {
169,081,118!
3177
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
3178
    return terrno;
×
3179
  }
3180
  uint64_t gid = pInfo->groupId;
169,081,118✔
3181

3182
  int32_t start = 0;
169,181,387✔
3183
  void*   tmp = taosArrayPush(pList, &start);
169,199,783✔
3184
  if (!tmp) {
169,199,783!
3185
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
3186
    return terrno;
×
3187
  }
3188

3189
  for (int32_t i = 1; i < size; ++i) {
766,839,971✔
3190
    pInfo = taosArrayGet(pTableListInfo->pTableList, i);
597,640,935✔
3191
    if (!pInfo) {
597,591,210!
3192
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
3193
      return terrno;
×
3194
    }
3195
    if (pInfo->groupId != gid) {
597,591,210✔
3196
      tmp = taosArrayPush(pList, &i);
62,957,014✔
3197
      if (!tmp) {
62,957,014!
3198
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
3199
        return terrno;
×
3200
      }
3201
      gid = pInfo->groupId;
62,957,014✔
3202
    }
3203
  }
3204

3205
  pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);
169,214,759✔
3206
  pTableListInfo->groupOffset = taosMemoryMalloc(sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
169,211,311✔
3207
  if (pTableListInfo->groupOffset == NULL) {
169,159,868!
3208
    taosArrayDestroy(pList);
×
3209
    return terrno;
×
3210
  }
3211

3212
  memcpy(pTableListInfo->groupOffset, taosArrayGet(pList, 0), sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
169,122,007!
3213
  taosArrayDestroy(pList);
169,170,382✔
3214
  return TSDB_CODE_SUCCESS;
169,056,779✔
3215
}
3216

3217
int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle* pHandle, SScanPhysiNode* pScanNode,
817,728,469✔
3218
                                    SNodeList* group, bool groupSort, uint8_t* digest, SStorageAPI* pAPI, SHashObj* groupIdMap) {
3219
  int32_t code = TSDB_CODE_SUCCESS;
817,728,469✔
3220

3221
  bool   groupByTbname = groupbyTbname(group);
817,728,469✔
3222
  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
817,767,442✔
3223
  if (!numOfTables) {
817,693,572!
3224
    return code;
×
3225
  }
3226
  qDebug("numOfTables:%zu, groupByTbname:%d, group:%p", numOfTables, groupByTbname, group);
817,693,572✔
3227
  if (group == NULL || groupByTbname) {
817,572,689✔
3228
    if (tsCountAlwaysReturnValue && QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode) &&
791,066,487✔
3229
        ((STableScanPhysiNode*)pScanNode)->needCountEmptyTable) {
563,761,412!
3230
      pTableListInfo->remainGroups =
68,209,396✔
3231
          taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
68,210,005✔
3232
      if (pTableListInfo->remainGroups == NULL) {
68,210,005!
3233
        return terrno;
×
3234
      }
3235

3236
      for (int i = 0; i < numOfTables; i++) {
394,212,459✔
3237
        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
326,003,316✔
3238
        if (!info) {
326,000,839!
3239
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
3240
          return terrno;
×
3241
        }
3242
        info->groupId = groupByTbname ? info->uid : 0;
326,000,839✔
3243
        int32_t tempRes = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId),
326,003,135✔
3244
                                      &(info->uid), sizeof(info->uid));
326,004,953✔
3245
        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
326,004,808!
3246
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
3247
          return tempRes;
×
3248
        }
3249
      }
3250
    } else {
3251
      for (int32_t i = 0; i < numOfTables; i++) {
2,147,483,647✔
3252
        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
2,147,483,647✔
3253
        if (!info) {
2,147,483,647!
3254
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
3255
          return terrno;
×
3256
        }
3257
        info->groupId = groupByTbname ? info->uid : 0;
2,147,483,647✔
3258
        
3259
      }
3260
    }
3261
    if (groupIdMap && group != NULL){
791,007,752✔
3262
      getColInfoResultForGroupbyForStream(pHandle->vnode, group, pTableListInfo, pAPI, groupIdMap);
1,907,725✔
3263
    }
3264

3265
    pTableListInfo->oneTableForEachGroup = groupByTbname;
791,007,752✔
3266
    if (numOfTables == 1 && pTableListInfo->idInfo.tableType == TSDB_CHILD_TABLE) {
791,229,695✔
3267
      pTableListInfo->oneTableForEachGroup = true;
151,134,337✔
3268
    }
3269

3270
    if (groupSort && groupByTbname) {
791,211,254✔
3271
      taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
16,036,837✔
3272
      pTableListInfo->numOfOuputGroups = numOfTables;
16,037,912✔
3273
    } else if (groupByTbname && pScanNode->groupOrderScan) {
775,174,417!
3274
      pTableListInfo->numOfOuputGroups = numOfTables;
161,120✔
3275
    } else {
3276
      pTableListInfo->numOfOuputGroups = 1;
775,013,571✔
3277
    }
3278
    if (groupSort || pScanNode->groupOrderScan) {
791,195,681!
3279
      code = sortTableGroup(pTableListInfo);
168,368,450✔
3280
    }
3281
  } else {
3282
    bool initRemainGroups = false;
26,506,202✔
3283
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {
26,506,202✔
3284
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;
26,138,147✔
3285
      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&
26,138,147!
3286
          !(groupSort || pScanNode->groupOrderScan)) {
12,239,361!
3287
        initRemainGroups = true;
12,173,013✔
3288
      }
3289
    }
3290

3291
    code = getColInfoResultForGroupby(pHandle->vnode, group, pTableListInfo, digest, pAPI, initRemainGroups, groupIdMap);
26,506,202✔
3292
    if (code != TSDB_CODE_SUCCESS) {
26,504,321✔
3293
      return code;
7,773✔
3294
    }
3295

3296
    if (pScanNode->groupOrderScan) pTableListInfo->numOfOuputGroups = taosArrayGetSize(pTableListInfo->pTableList);
26,496,548!
3297

3298
    if (groupSort || pScanNode->groupOrderScan) {
26,497,157!
3299
      code = sortTableGroup(pTableListInfo);
819,260✔
3300
    }
3301
  }
3302

3303
  // add all table entry in the hash map
3304
  size_t size = taosArrayGetSize(pTableListInfo->pTableList);
817,546,838✔
3305
  for (int32_t i = 0; i < size; ++i) {
2,147,483,647✔
3306
    STableKeyInfo* p = taosArrayGet(pTableListInfo->pTableList, i);
2,147,483,647✔
3307
    if (!p) {
2,147,483,647!
3308
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
3309
      return terrno;
×
3310
    }
3311
    int32_t tempRes = taosHashPut(pTableListInfo->map, &p->uid, sizeof(uint64_t), &i, sizeof(int32_t));
2,147,483,647✔
3312
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
2,147,483,647!
3313
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
3314
      return tempRes;
×
3315
    }
3316
  }
3317

3318
  return code;
817,836,250✔
3319
}
3320

3321
int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
855,568,060✔
3322
                                STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
3323
                                SExecTaskInfo* pTaskInfo, SHashObj* groupIdMap) {
3324
  int64_t     st = taosGetTimestampUs();
855,629,881✔
3325
  const char* idStr = GET_TASKID(pTaskInfo);
855,629,881✔
3326

3327
  if (pHandle == NULL) {
854,812,052!
3328
    qError("invalid handle, in creating operator tree, %s", idStr);
×
3329
    return TSDB_CODE_INVALID_PARA;
×
3330
  }
3331

3332
  if (pHandle->uid != 0) {
854,812,052✔
3333
    pScanNode->uid = pHandle->uid;
120,439✔
3334
    pScanNode->tableType = TSDB_CHILD_TABLE;
120,439✔
3335
  }
3336
  uint8_t digest[17] = {0};
855,385,367✔
3337
  int32_t code = getTableList(pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, digest, idStr,
855,340,005✔
3338
                              &pTaskInfo->storageAPI, pTaskInfo->pStreamRuntimeInfo);
855,553,725✔
3339
  if (code != TSDB_CODE_SUCCESS) {
855,718,430✔
3340
    qError("failed to getTableList, code:%s", tstrerror(code));
4,668!
3341
    return code;
4,668✔
3342
  }
3343

3344
  int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
855,713,762✔
3345

3346
  int64_t st1 = taosGetTimestampUs();
855,789,171✔
3347
  pTaskInfo->cost.extractListTime = (st1 - st) / 1000.0;
855,789,171✔
3348
  qDebug("extract queried table list completed, %d tables, elapsed time:%.2f ms %s", numOfTables,
855,855,872✔
3349
         pTaskInfo->cost.extractListTime, idStr);
3350

3351
  if (numOfTables == 0) {
855,745,855✔
3352
    qDebug("no table qualified for query, %s", idStr);
38,034,993✔
3353
    return TSDB_CODE_SUCCESS;
38,033,907✔
3354
  }
3355

3356
  code = buildGroupIdMapForAllTables(pTableListInfo, pHandle, pScanNode, pGroupTags, groupSort, digest, &pTaskInfo->storageAPI, groupIdMap);
817,710,862✔
3357
  if (code != TSDB_CODE_SUCCESS) {
817,827,197✔
3358
    return code;
7,773✔
3359
  }
3360

3361
  pTaskInfo->cost.groupIdMapTime = (taosGetTimestampUs() - st1) / 1000.0;
817,822,883✔
3362
  qDebug("generate group id map completed, elapsed time:%.2f ms %s", pTaskInfo->cost.groupIdMapTime, idStr);
817,810,350✔
3363

3364
  return TSDB_CODE_SUCCESS;
817,784,507✔
3365
}
3366

3367
char* getStreamOpName(uint16_t opType) {
31,741,340✔
3368
  switch (opType) {
31,741,340!
3369
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
×
3370
      return "stream scan";
×
3371
    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
31,372,040✔
3372
      return "project";
31,372,040✔
3373
    case QUERY_NODE_PHYSICAL_PLAN_EXTERNAL_WINDOW:
369,300✔
3374
      return "external window";
369,300✔
3375
  }
3376
  return "error name";
×
3377
}
3378

3379
void printDataBlock(SSDataBlock* pBlock, const char* flag, const char* taskIdStr, int64_t qId) {
2,007,168,964✔
3380
  if (qDebugFlag & DEBUG_DEBUG) {
2,007,168,964✔
3381
    if (!pBlock) {
1,875,476,677✔
3382
      qDebug("%" PRIx64 " %s %s %s: Block is Null", qId, taskIdStr, flag, __func__);
38,867,696!
3383
      return;
38,871,381✔
3384
    } else if (pBlock->info.rows == 0) {
1,836,608,981✔
3385
      qDebug("%" PRIx64 " %s %s %s: Block is Empty. block type %d", qId, taskIdStr, flag, __func__, pBlock->info.type);
139,678,429✔
3386
      return;
139,680,712✔
3387
    }
3388
    
3389
    char*   pBuf = NULL;
1,696,886,480✔
3390
    int32_t code = dumpBlockData(pBlock, flag, &pBuf, taskIdStr, qId);
1,696,894,009✔
3391
    if (code == 0) {
1,696,949,625!
3392
      qDebugL("%" PRIx64 " %s %s", qId, __func__, pBuf);
1,696,972,833!
3393
      taosMemoryFree(pBuf);
1,696,973,319✔
3394
    }
3395
  }
3396
}
3397

3398
void printSpecDataBlock(SSDataBlock* pBlock, const char* flag, const char* opStr, const char* taskIdStr) {
×
3399
  if (!pBlock) {
×
3400
    qDebug("%s===stream===%s %s: Block is Null", taskIdStr, flag, opStr);
×
3401
    return;
×
3402
  } else if (pBlock->info.rows == 0) {
×
3403
    qDebug("%s===stream===%s %s: Block is Empty. block type %d.skey:%" PRId64 ",ekey:%" PRId64 ",version%" PRId64,
×
3404
           taskIdStr, flag, opStr, pBlock->info.type, pBlock->info.window.skey, pBlock->info.window.ekey,
3405
           pBlock->info.version);
3406
    return;
×
3407
  }
3408
  if (qDebugFlag & DEBUG_DEBUG) {
×
3409
    char* pBuf = NULL;
×
3410
    char  flagBuf[64];
×
3411
    snprintf(flagBuf, sizeof(flagBuf), "%s %s", flag, opStr);
×
3412
    int32_t code = dumpBlockData(pBlock, flagBuf, &pBuf, taskIdStr, 0);
×
3413
    if (code == 0) {
×
3414
      qDebug("%s", pBuf);
×
3415
      taosMemoryFree(pBuf);
×
3416
    }
3417
  }
3418
}
3419

3420
TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }
928,611,847!
3421

3422
void updateTimeWindowInfo(SColumnInfoData* pColData, const STimeWindow* pWin, int64_t delta) {
2,147,483,647✔
3423
  int64_t* ts = (int64_t*)pColData->pData;
2,147,483,647✔
3424

3425
  int64_t duration = pWin->ekey - pWin->skey + delta;
2,147,483,647✔
3426
  ts[2] = duration;            // set the duration
2,147,483,647✔
3427
  ts[3] = pWin->skey;          // window start key
2,147,483,647✔
3428
  ts[4] = pWin->ekey + delta;  // window end key
2,147,483,647✔
3429
}
2,147,483,647✔
3430

3431
int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t oldKeysLen, const SSDataBlock* pBlock,
2,147,483,647✔
3432
                 int32_t rowIndex) {
3433
  SColumnDataAgg* pColAgg = NULL;
2,147,483,647✔
3434
  const char*     isNull = oldkeyBuf;
2,147,483,647✔
3435
  const char*     p = oldkeyBuf + sizeof(int8_t) * pSortGroupCols->size;
2,147,483,647✔
3436

3437
  for (int32_t i = 0; i < pSortGroupCols->size; ++i) {
2,147,483,647✔
3438
    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
2,147,483,647✔
3439
    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
2,147,483,647✔
3440
    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
2,147,483,647!
3441

3442
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
2,147,483,647!
3443
      if (isNull[i] != 1) return 1;
2,147,483,647✔
3444
    } else {
3445
      if (isNull[i] != 0) return 1;
2,147,483,647✔
3446
      const char* val = colDataGetData(pColInfoData, rowIndex);
2,147,483,647!
3447
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
2,147,483,647!
3448
        int32_t len = getJsonValueLen(val);
×
3449
        if (memcmp(p, val, len) != 0) return 1;
×
3450
        p += len;
×
3451
      } else if (IS_VAR_DATA_TYPE(pCol->type)) {
2,147,483,647!
3452
        if (IS_STR_DATA_BLOB(pCol->type)) {
2,147,483,647!
UNCOV
3453
          if (memcmp(p, val, blobDataTLen(val)) != 0) return 1;
×
3454
          p += blobDataTLen(val);
×
3455
        } else {
3456
          if (memcmp(p, val, varDataTLen(val)) != 0) return 1;
2,147,483,647!
3457
          p += varDataTLen(val);
2,147,483,647✔
3458
        }
3459
      } else {
3460
        if (0 != memcmp(p, val, pCol->bytes)) return 1;
2,147,483,647!
3461
        p += pCol->bytes;
2,147,483,647✔
3462
      }
3463
    }
3464
  }
3465
  if ((int32_t)(p - oldkeyBuf) != oldKeysLen) return 1;
2,147,483,647✔
3466
  return 0;
2,147,483,647✔
3467
}
3468

3469
int32_t buildKeys(char* keyBuf, const SArray* pSortGroupCols, const SSDataBlock* pBlock, int32_t rowIndex) {
498,158,088✔
3470
  uint32_t        colNum = pSortGroupCols->size;
498,158,088✔
3471
  SColumnDataAgg* pColAgg = NULL;
498,273,770✔
3472
  char*           isNull = keyBuf;
498,273,770✔
3473
  char*           p = keyBuf + sizeof(int8_t) * colNum;
498,273,770✔
3474

3475
  for (int32_t i = 0; i < colNum; ++i) {
1,487,765,844✔
3476
    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
989,456,346✔
3477
    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
989,481,130✔
3478
    if (pCol->slotId > pBlock->pDataBlock->size) continue;
989,521,372!
3479

3480
    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
989,537,265!
3481

3482
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
1,979,030,249!
3483
      isNull[i] = 1;
35,554,730✔
3484
    } else {
3485
      isNull[i] = 0;
953,988,216✔
3486
      const char* val = colDataGetData(pColInfoData, rowIndex);
953,932,674!
3487
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
953,930,162!
3488
        int32_t len = getJsonValueLen(val);
×
3489
        memcpy(p, val, len);
×
3490
        p += len;
×
3491
      } else if (IS_VAR_DATA_TYPE(pCol->type)) {
953,849,840!
3492
        if (IS_STR_DATA_BLOB(pCol->type)) {
145,271,290!
3493
          blobDataCopy(p, val);
332!
3494
          p += blobDataTLen(val);
×
3495
        } else {
3496
          varDataCopy(p, val);
145,173,273!
3497
          p += varDataTLen(val);
145,178,128✔
3498
        }
3499
      } else {
3500
        memcpy(p, val, pCol->bytes);
808,710,284!
3501
        p += pCol->bytes;
808,697,649✔
3502
      }
3503
    }
3504
  }
3505
  return (int32_t)(p - keyBuf);
498,309,498✔
3506
}
3507

3508
uint64_t calcGroupId(char* pData, int32_t len) {
2,147,483,647✔
3509
  T_MD5_CTX context;
2,147,483,647✔
3510
  tMD5Init(&context);
2,147,483,647✔
3511
  tMD5Update(&context, (uint8_t*)pData, len);
2,147,483,647✔
3512
  tMD5Final(&context);
2,147,483,647✔
3513

3514
  // NOTE: only extract the initial 8 bytes of the final MD5 digest
3515
  uint64_t id = 0;
2,147,483,647✔
3516
  memcpy(&id, context.digest, sizeof(uint64_t));
2,147,483,647✔
3517
  if (0 == id) memcpy(&id, context.digest + 8, sizeof(uint64_t));
2,147,483,647!
3518
  return id;
2,147,483,647✔
3519
}
3520

3521
SNodeList* makeColsNodeArrFromSortKeys(SNodeList* pSortKeys) {
1,074,928✔
3522
  SNode*     node;
3523
  SNodeList* ret = NULL;
1,074,928✔
3524
  FOREACH(node, pSortKeys) {
3,281,533!
3525
    SOrderByExprNode* pSortKey = (SOrderByExprNode*)node;
2,207,312✔
3526
    int32_t           code = nodesListMakeAppend(&ret, pSortKey->pExpr);
2,207,312✔
3527
    if (code != TSDB_CODE_SUCCESS) {
2,206,605!
3528
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
3529
      terrno = code;
×
3530
      return NULL;
×
3531
    }
3532
  }
3533
  return ret;
1,074,928✔
3534
}
3535

3536
int32_t extractKeysLen(const SArray* keys, int32_t* pLen) {
1,073,872✔
3537
  int32_t code = TSDB_CODE_SUCCESS;
1,073,872✔
3538
  int32_t lino = 0;
1,073,872✔
3539
  int32_t len = 0;
1,073,872✔
3540
  int32_t keyNum = taosArrayGetSize(keys);
1,073,872✔
3541
  for (int32_t i = 0; i < keyNum; ++i) {
2,724,087✔
3542
    SColumn* pCol = (SColumn*)taosArrayGet(keys, i);
1,650,922✔
3543
    QUERY_CHECK_NULL(pCol, code, lino, _end, terrno);
1,650,922!
3544
    len += pCol->bytes;
1,650,922✔
3545
  }
3546
  len += sizeof(int8_t) * keyNum;  // null flag
1,073,165✔
3547
  *pLen = len;
1,073,165✔
3548

3549
_end:
1,073,165✔
3550
  if (code != TSDB_CODE_SUCCESS) {
1,073,165!
3551
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
3552
  }
3553
  return code;
1,074,221✔
3554
}
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