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

taosdata / TDengine / #4476

08 Jul 2025 09:38AM UTC coverage: 62.948% (+0.03%) from 62.922%
#4476

push

travis-ci

web-flow
Merge pull request #31712 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

158623 of 321496 branches covered (49.34%)

Branch coverage included in aggregate %.

56 of 60 new or added lines in 13 files covered. (93.33%)

1176 existing lines in 42 files now uncovered.

245593 of 320647 relevant lines covered (76.59%)

23441503.11 hits per line

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

68.52
/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);
59

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

68
void initResultRowInfo(SResultRowInfo* pResultRowInfo) {
21,040,736✔
69
  pResultRowInfo->size = 0;
21,040,736✔
70
  pResultRowInfo->cur.pageId = -1;
21,040,736✔
71
}
21,040,736✔
72

73
void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }
17,619,950✔
74

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

81
  if (entrySize > 0) {
115,674,589!
82
    memset(pResultRow->pEntryInfo, 0, entrySize);
115,692,956✔
83
  }
84
}
115,674,589✔
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) {
12,810,152✔
92
  int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow);
12,810,152✔
93

94
  for (int32_t i = 0; i < numOfOutput; ++i) {
54,668,088✔
95
    rowSize += pCtx[i].resDataInfo.interBufSize;
41,857,936✔
96
  }
97

98
  return rowSize;
12,810,152✔
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) {
3,571✔
103
  if (inBuf == NULL || pSup == NULL) {
3,571!
104
    qError("invalid input parameters, inBuf:%p, pSup:%p", inBuf, pSup);
9!
105
    return TSDB_CODE_INVALID_PARA;
9✔
106
  }
107
  SqlFunctionCtx* pCtx = pSup->pCtx;
3,562✔
108
  int32_t*        offset = pSup->rowEntryInfoOffset;
3,562✔
109
  SResultRow*     pResultRow = NULL;
3,562✔
110
  size_t          processedSize = 0;
3,562✔
111
  int32_t         code = TSDB_CODE_SUCCESS;
3,562✔
112

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

125
  for (int32_t i = 0; i < pSup->numOfExprs; ++i) {
64,226✔
126
    int32_t len = *(int32_t*)inBuf;
60,666✔
127
    inBuf += sizeof(int32_t);
60,666✔
128
    processedSize += sizeof(int32_t);
60,666✔
129
    if (pResultRow->version != FUNCTION_RESULT_INFO_VERSION && pCtx->fpSet.decode) {
60,666!
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);
60,666✔
137
    }
138
    inBuf += len;
60,664✔
139
    processedSize += len;
60,664✔
140
  }
141

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

156
  qTrace("[StreamInternal] get result inBufSize:%zu, outBufSize:%zu", inBufSize, *outBufSize);
3,560!
157
  return TSDB_CODE_SUCCESS;
3,562✔
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) {
431,639✔
162
  if (pSup == NULL || inBuf == NULL || outBuf == NULL || outBufSize == NULL) {
431,639!
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;
431,639✔
168
  int32_t*        offset = pSup->rowEntryInfoOffset;
431,639✔
169
  SResultRow*     pResultRow = (SResultRow*)inBuf;
431,639✔
170
  size_t          rowSize = getResultRowSize(pCtx, pSup->numOfExprs);
431,639✔
171

172
  if (rowSize > inBufSize) {
431,639!
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;
431,639✔
179
  if (rowSize < inBufSize) {
431,639✔
180
    *outBufSize += inBufSize - rowSize;
1,661✔
181
  }
182

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

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

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

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

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

214
void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
8,494,258✔
215
  taosMemoryFreeClear(pGroupResInfo->pBuf);
8,494,258!
216
  if (pGroupResInfo->freeItem) {
8,495,905!
217
    //    taosArrayDestroy(pGroupResInfo->pRows);
218
    taosArrayDestroyEx(pGroupResInfo->pRows, freeEx);
×
219
    pGroupResInfo->freeItem = false;
×
220
    pGroupResInfo->pRows = NULL;
×
221
  } else {
222
    taosArrayDestroy(pGroupResInfo->pRows);
8,495,905✔
223
    pGroupResInfo->pRows = NULL;
8,496,957✔
224
  }
225
  pGroupResInfo->index = 0;
8,496,957✔
226
  pGroupResInfo->delIndex = 0;
8,496,957✔
227
}
8,496,957✔
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;
190,546,221✔
244
  }
245
}
246

247
static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); }
92,163,662✔
248

249
int32_t initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order) {
6,856,072✔
250
  int32_t code = TSDB_CODE_SUCCESS;
6,856,072✔
251
  int32_t lino = 0;
6,856,072✔
252
  if (pGroupResInfo->pRows != NULL) {
6,856,072✔
253
    taosArrayDestroy(pGroupResInfo->pRows);
50,652✔
254
  }
255
  if (pGroupResInfo->pBuf) {
6,856,073✔
256
    taosMemoryFree(pGroupResInfo->pBuf);
50,653!
257
    pGroupResInfo->pBuf = NULL;
50,653✔
258
  }
259

260
  // extract the result rows information from the hash map
261
  int32_t size = tSimpleHashGetSize(pHashmap);
6,856,073✔
262

263
  void* pData = NULL;
6,856,809✔
264
  pGroupResInfo->pRows = taosArrayInit(size, POINTER_BYTES);
6,856,809✔
265
  QUERY_CHECK_NULL(pGroupResInfo->pRows, code, lino, _end, terrno);
6,857,814!
266

267
  size_t  keyLen = 0;
6,857,814✔
268
  int32_t iter = 0;
6,857,814✔
269
  int64_t bufLen = 0, offset = 0;
6,857,814✔
270

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

277
  pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);
6,838,565!
278
  QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
6,858,041!
279

280
  iter = 0;
6,858,041✔
281
  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
338,153,363✔
282
    void* key = tSimpleHashGetKey(pData, &keyLen);
331,333,022✔
283

284
    SResKeyPos* p = (SResKeyPos*)(pGroupResInfo->pBuf + offset);
331,333,022✔
285

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

292
    offset += keyLen + sizeof(struct SResultRowPosition);
331,295,322✔
293
  }
294

295
  if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
6,857,545✔
296
    __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc;
3,122,480✔
297
    size = POINTER_BYTES;
3,122,480✔
298
    taosSort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), size, fn);
3,122,480✔
299
  }
300

301
  pGroupResInfo->index = 0;
6,859,123✔
302

303
_end:
6,859,123✔
304
  if (code != TSDB_CODE_SUCCESS) {
6,859,123!
305
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
306
  }
307
  return code;
6,857,855✔
308
}
309

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

315
  pGroupResInfo->freeItem = true;
28,581✔
316
  pGroupResInfo->pRows = pArrayList;
28,581✔
317
  pGroupResInfo->index = 0;
28,581✔
318
  pGroupResInfo->delIndex = 0;
28,581✔
319
}
28,581✔
320

321
bool hasRemainResults(SGroupResInfo* pGroupResInfo) {
19,142,598✔
322
  if (pGroupResInfo->pRows == NULL) {
19,142,598✔
323
    return false;
2,106✔
324
  }
325

326
  return pGroupResInfo->index < taosArrayGetSize(pGroupResInfo->pRows);
19,140,492✔
327
}
328

329
int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {
12,776,162✔
330
  if (pGroupResInfo->pRows == 0) {
12,776,162✔
331
    return 0;
3,816✔
332
  }
333

334
  return (int32_t)taosArrayGetSize(pGroupResInfo->pRows);
12,772,346✔
335
}
336

337
SArray* createSortInfo(SNodeList* pNodeList) {
4,155,898✔
338
  size_t numOfCols = 0;
4,155,898✔
339

340
  if (pNodeList != NULL) {
4,155,898✔
341
    numOfCols = LIST_LENGTH(pNodeList);
4,152,925!
342
  } else {
343
    numOfCols = 0;
2,973✔
344
  }
345

346
  SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));
4,155,898✔
347
  if (pList == NULL) {
4,156,014!
348
    return pList;
×
349
  }
350

351
  for (int32_t i = 0; i < numOfCols; ++i) {
8,529,212✔
352
    SOrderByExprNode* pSortKey = (SOrderByExprNode*)nodesListGetNode(pNodeList, i);
4,373,099✔
353
    if (!pSortKey) {
4,373,112!
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};
4,373,135✔
361
    bi.order = (pSortKey->order == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
4,373,135✔
362
    bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST);
4,373,135✔
363

364
    SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr;
4,373,135✔
365
    bi.slotId = pColNode->slotId;
4,373,135✔
366
    void* tmp = taosArrayPush(pList, &bi);
4,373,198✔
367
    if (!tmp) {
4,373,198!
368
      taosArrayDestroy(pList);
×
369
      pList = NULL;
×
370
      break;
×
371
    }
372
  }
373

374
  return pList;
4,156,113✔
375
}
376

377
SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {
37,931,380✔
378
  int32_t      numOfCols = LIST_LENGTH(pNode->pSlots);
37,931,380!
379
  SSDataBlock* pBlock = NULL;
37,931,380✔
380
  int32_t      code = createDataBlock(&pBlock);
37,931,380✔
381
  if (code) {
37,937,996!
382
    terrno = code;
×
383
    return NULL;
×
384
  }
385

386
  pBlock->info.id.blockId = pNode->dataBlockId;
37,944,938✔
387
  pBlock->info.type = STREAM_INVALID;
37,944,938✔
388
  pBlock->info.calWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
37,944,938✔
389
  pBlock->info.watermark = INT64_MIN;
37,944,938✔
390

391
  for (int32_t i = 0; i < numOfCols; ++i) {
206,974,207✔
392
    SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
168,993,106✔
393
    if (!pDescNode) {
168,900,613!
394
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
395
      blockDataDestroy(pBlock);
×
396
      pBlock = NULL;
×
397
      terrno = TSDB_CODE_INVALID_PARA;
×
398
      break;
×
399
    }
400
    SColumnInfoData idata =
401
        createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);
168,900,613✔
402
    idata.info.scale = pDescNode->dataType.scale;
169,333,922✔
403
    idata.info.precision = pDescNode->dataType.precision;
169,333,922✔
404
    idata.info.noData = pDescNode->reserve;
169,333,922✔
405

406
    code = blockDataAppendColInfo(pBlock, &idata);
169,333,922✔
407
    if (code != TSDB_CODE_SUCCESS) {
169,029,269!
408
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
409
      blockDataDestroy(pBlock);
×
410
      pBlock = NULL;
×
411
      terrno = code;
×
412
      break;
×
413
    }
414
  }
415

416
  return pBlock;
37,981,101✔
417
}
418

419
int32_t prepareDataBlockBuf(SSDataBlock* pDataBlock, SColMatchInfo* pMatchInfo) {
8,555,909✔
420
  SDataBlockInfo* pBlockInfo = &pDataBlock->info;
8,555,909✔
421

422
  for (int32_t i = 0; i < taosArrayGetSize(pMatchInfo->pList); ++i) {
52,814,166✔
423
    SColMatchItem* pItem = taosArrayGet(pMatchInfo->pList, i);
44,621,701✔
424
    if (!pItem) {
44,658,115!
UNCOV
425
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
UNCOV
426
      return terrno;
×
427
    }
428

429
    if (pItem->isPk) {
44,658,249✔
430
      SColumnInfoData* pInfoData = taosArrayGet(pDataBlock->pDataBlock, pItem->dstSlotId);
399,992✔
431
      if (!pInfoData) {
400,009!
432
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
433
        return terrno;
×
434
      }
435
      pBlockInfo->pks[0].type = pInfoData->info.type;
400,009✔
436
      pBlockInfo->pks[1].type = pInfoData->info.type;
400,009✔
437

438
      // allocate enough buffer size, which is pInfoData->info.bytes
439
      if (IS_VAR_DATA_TYPE(pItem->dataType.type)) {
400,009!
440
        pBlockInfo->pks[0].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
35,511!
441
        if (pBlockInfo->pks[0].pData == NULL) {
35,437!
442
          return terrno;
×
443
        }
444

445
        pBlockInfo->pks[1].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
35,437!
446
        if (pBlockInfo->pks[1].pData == NULL) {
35,447!
447
          taosMemoryFreeClear(pBlockInfo->pks[0].pData);
×
448
          return terrno;
×
449
        }
450

451
        pBlockInfo->pks[0].nData = pInfoData->info.bytes;
35,447✔
452
        pBlockInfo->pks[1].nData = pInfoData->info.bytes;
35,447✔
453
      }
454

455
      break;
399,945✔
456
    }
457
  }
458

459
  return TSDB_CODE_SUCCESS;
8,529,351✔
460
}
461

462
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
6,490✔
463
  STransTagExprCtx* pCtx = pContext;
6,490✔
464
  SMetaReader*      mr = pCtx->pReader;
6,490✔
465
  bool              isTagCol = false, isTbname = false;
6,490✔
466
  if (nodeType(*pNode) == QUERY_NODE_COLUMN) {
6,490✔
467
    SColumnNode* pCol = (SColumnNode*)*pNode;
1,873✔
468
    if (pCol->colType == COLUMN_TYPE_TBNAME)
1,873!
469
      isTbname = true;
×
470
    else
471
      isTagCol = true;
1,873✔
472
  } else if (nodeType(*pNode) == QUERY_NODE_FUNCTION) {
4,617!
473
    SFunctionNode* pFunc = (SFunctionNode*)*pNode;
×
474
    if (pFunc->funcType == FUNCTION_TYPE_TBNAME) isTbname = true;
×
475
  }
476
  if (isTagCol) {
6,490✔
477
    SColumnNode* pSColumnNode = *(SColumnNode**)pNode;
1,873✔
478

479
    SValueNode* res = NULL;
1,873✔
480
    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
1,873✔
481
    if (NULL == res) {
1,873!
482
      return DEAL_RES_ERROR;
×
483
    }
484

485
    res->translate = true;
1,873✔
486
    res->node.resType = pSColumnNode->node.resType;
1,873✔
487

488
    STagVal tagVal = {0};
1,873✔
489
    tagVal.cid = pSColumnNode->colId;
1,873✔
490
    const char* p = mr->pAPI->extractTagVal(mr->me.ctbEntry.pTags, pSColumnNode->node.resType.type, &tagVal);
1,873✔
491
    if (p == NULL) {
1,873!
492
      res->node.resType.type = TSDB_DATA_TYPE_NULL;
×
493
    } else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {
1,873!
494
      int32_t len = ((const STag*)p)->len;
×
495
      res->datum.p = taosMemoryCalloc(len + 1, 1);
×
496
      if (NULL == res->datum.p) {
×
497
        return DEAL_RES_ERROR;
×
498
      }
499
      memcpy(res->datum.p, p, len);
×
500
    } else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) {
1,873!
501
      res->datum.p = taosMemoryCalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1, 1);
1,842!
502
      if (NULL == res->datum.p) {
1,842!
503
        return DEAL_RES_ERROR;
×
504
      }
505
      memcpy(varDataVal(res->datum.p), tagVal.pData, tagVal.nData);
1,842✔
506
      varDataSetLen(res->datum.p, tagVal.nData);
1,842✔
507
    } else {
508
      int32_t code = nodesSetValueNodeValue(res, &(tagVal.i64));
31✔
509
      if (code != TSDB_CODE_SUCCESS) {
31!
510
        return DEAL_RES_ERROR;
×
511
      }
512
    }
513
    nodesDestroyNode(*pNode);
1,873✔
514
    *pNode = (SNode*)res;
1,873✔
515
  } else if (isTbname) {
4,617!
516
    SValueNode* res = NULL;
×
517
    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
×
518
    if (NULL == res) {
×
519
      return DEAL_RES_ERROR;
×
520
    }
521

522
    res->translate = true;
×
523
    res->node.resType = ((SExprNode*)(*pNode))->resType;
×
524

525
    int32_t len = strlen(mr->me.name);
×
526
    res->datum.p = taosMemoryCalloc(len + VARSTR_HEADER_SIZE + 1, 1);
×
527
    if (NULL == res->datum.p) {
×
528
      return DEAL_RES_ERROR;
×
529
    }
530
    memcpy(varDataVal(res->datum.p), mr->me.name, len);
×
531
    varDataSetLen(res->datum.p, len);
×
532
    nodesDestroyNode(*pNode);
×
533
    *pNode = (SNode*)res;
×
534
  }
535

536
  return DEAL_RES_CONTINUE;
6,490✔
537
}
538

539
int32_t isQualifiedTable(STableKeyInfo* info, SNode* pTagCond, void* metaHandle, bool* pQualified, SStorageAPI* pAPI) {
930✔
540
  int32_t     code = TSDB_CODE_SUCCESS;
930✔
541
  SMetaReader mr = {0};
930✔
542

543
  pAPI->metaReaderFn.initReader(&mr, metaHandle, META_READER_LOCK, &pAPI->metaFn);
930✔
544
  code = pAPI->metaReaderFn.getEntryGetUidCache(&mr, info->uid);
930✔
545
  if (TSDB_CODE_SUCCESS != code) {
930!
546
    pAPI->metaReaderFn.clearReader(&mr);
×
547
    *pQualified = false;
×
548

549
    return TSDB_CODE_SUCCESS;
×
550
  }
551

552
  SNode* pTagCondTmp = NULL;
930✔
553
  code = nodesCloneNode(pTagCond, &pTagCondTmp);
930✔
554
  if (TSDB_CODE_SUCCESS != code) {
930!
555
    *pQualified = false;
×
556
    pAPI->metaReaderFn.clearReader(&mr);
×
557
    return code;
×
558
  }
559
  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
930✔
560
  nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &ctx);
930✔
561
  pAPI->metaReaderFn.clearReader(&mr);
930✔
562
  if (TSDB_CODE_SUCCESS != ctx.code) {
930!
563
    *pQualified = false;
×
564
    terrno = code;
×
565
    return code;
×
566
  }
567

568
  SNode* pNew = NULL;
930✔
569
  code = scalarCalculateConstants(pTagCondTmp, &pNew);
930✔
570
  if (TSDB_CODE_SUCCESS != code) {
930!
571
    terrno = code;
×
572
    nodesDestroyNode(pTagCondTmp);
×
573
    *pQualified = false;
×
574

575
    return code;
×
576
  }
577

578
  SValueNode* pValue = (SValueNode*)pNew;
930✔
579
  *pQualified = pValue->datum.b;
930✔
580

581
  nodesDestroyNode(pNew);
930✔
582
  return TSDB_CODE_SUCCESS;
930✔
583
}
584

585
static EDealRes getColumn(SNode** pNode, void* pContext) {
147,795,218✔
586
  tagFilterAssist* pData = (tagFilterAssist*)pContext;
147,795,218✔
587
  SColumnNode*     pSColumnNode = NULL;
147,795,218✔
588
  if (QUERY_NODE_COLUMN == nodeType((*pNode))) {
147,795,218✔
589
    pSColumnNode = *(SColumnNode**)pNode;
6,852,668✔
590
  } else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {
140,942,550✔
591
    SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);
857,056✔
592
    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {
857,056✔
593
      pData->code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pSColumnNode);
855,895✔
594
      if (NULL == pSColumnNode) {
855,889!
595
        return DEAL_RES_ERROR;
×
596
      }
597
      pSColumnNode->colId = -1;
855,889✔
598
      pSColumnNode->colType = COLUMN_TYPE_TBNAME;
855,889✔
599
      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;
855,889✔
600
      pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;
855,889✔
601
      nodesDestroyNode(*pNode);
855,889✔
602
      *pNode = (SNode*)pSColumnNode;
855,901✔
603
    } else {
604
      return DEAL_RES_CONTINUE;
1,161✔
605
    }
606
  } else {
607
    return DEAL_RES_CONTINUE;
140,085,494✔
608
  }
609

610
  void* data = taosHashGet(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));
7,708,569✔
611
  if (!data) {
7,703,021✔
612
    int32_t tempRes =
613
        taosHashPut(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));
6,126,712✔
614
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
6,133,175!
615
      return DEAL_RES_ERROR;
×
616
    }
617
    pSColumnNode->slotId = pData->index++;
6,133,175✔
618
    SColumnInfo cInfo = {.colId = pSColumnNode->colId,
6,133,175✔
619
                         .type = pSColumnNode->node.resType.type,
6,133,175✔
620
                         .bytes = pSColumnNode->node.resType.bytes,
6,133,175✔
621
                         .pk = pSColumnNode->isPk};
6,133,175✔
622
#if TAG_FILTER_DEBUG
623
    qDebug("tagfilter build column info, slotId:%d, colId:%d, type:%d", pSColumnNode->slotId, cInfo.colId, cInfo.type);
624
#endif
625
    void* tmp = taosArrayPush(pData->cInfoList, &cInfo);
6,133,175✔
626
    if (!tmp) {
6,131,016!
627
      return DEAL_RES_ERROR;
×
628
    }
629
  } else {
630
    SColumnNode* col = *(SColumnNode**)data;
1,576,309✔
631
    pSColumnNode->slotId = col->slotId;
1,576,309✔
632
  }
633

634
  return DEAL_RES_CONTINUE;
7,707,325✔
635
}
636

637
static int32_t createResultData(SDataType* pType, int32_t numOfRows, SScalarParam* pParam) {
2,453,847✔
638
  SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
2,453,847!
639
  if (pColumnData == NULL) {
2,454,756!
640
    return terrno;
×
641
  }
642

643
  pColumnData->info.type = pType->type;
2,454,756✔
644
  pColumnData->info.bytes = pType->bytes;
2,454,756✔
645
  pColumnData->info.scale = pType->scale;
2,454,756✔
646
  pColumnData->info.precision = pType->precision;
2,454,756✔
647

648
  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);
2,454,756✔
649
  if (code != TSDB_CODE_SUCCESS) {
2,454,597!
650
    terrno = code;
×
651
    releaseColInfoData(pColumnData);
×
652
    return terrno;
×
653
  }
654

655
  pParam->columnData = pColumnData;
2,454,662✔
656
  pParam->colAlloced = true;
2,454,662✔
657
  return TSDB_CODE_SUCCESS;
2,454,662✔
658
}
659

660
static void releaseColInfoData(void* pCol) {
66,354✔
661
  if (pCol) {
66,354!
662
    SColumnInfoData* col = (SColumnInfoData*)pCol;
66,359✔
663
    colDataDestroy(col);
66,359✔
664
    taosMemoryFree(col);
66,366!
665
  }
666
}
66,365✔
667

668
void freeItem(void* p) {
5,599,965✔
669
  STUidTagInfo* pInfo = p;
5,599,965✔
670
  if (pInfo->pTagVal != NULL) {
5,599,965✔
671
    taosMemoryFree(pInfo->pTagVal);
5,255,468!
672
  }
673
}
5,600,442✔
674

675
static int32_t genTagFilterDigest(const SNode* pTagCond, T_MD5_CTX* pContext) {
×
676
  if (pTagCond == NULL) {
×
677
    return TSDB_CODE_SUCCESS;
×
678
  }
679

680
  char*   payload = NULL;
×
681
  int32_t len = 0;
×
682
  int32_t code = nodesNodeToMsg(pTagCond, &payload, &len);
×
683
  if (code != TSDB_CODE_SUCCESS) {
×
684
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
685
    return code;
×
686
  }
687

688
  tMD5Init(pContext);
×
689
  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
×
690
  tMD5Final(pContext);
×
691

692
  taosMemoryFree(payload);
×
693
  return TSDB_CODE_SUCCESS;
×
694
}
695

696
static int32_t genTbGroupDigest(const SNode* pGroup, uint8_t* filterDigest, T_MD5_CTX* pContext) {
×
697
  int32_t code = TSDB_CODE_SUCCESS;
×
698
  int32_t lino = 0;
×
699
  char*   payload = NULL;
×
700
  int32_t len = 0;
×
701
  code = nodesNodeToMsg(pGroup, &payload, &len);
×
702
  QUERY_CHECK_CODE(code, lino, _end);
×
703

704
  if (filterDigest[0]) {
×
705
    payload = taosMemoryRealloc(payload, len + tListLen(pContext->digest));
×
706
    QUERY_CHECK_NULL(payload, code, lino, _end, terrno);
×
707
    memcpy(payload + len, filterDigest + 1, tListLen(pContext->digest));
×
708
    len += tListLen(pContext->digest);
×
709
  }
710

711
  tMD5Init(pContext);
×
712
  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
×
713
  tMD5Final(pContext);
×
714

715
_end:
×
716
  taosMemoryFree(payload);
×
717
  if (code != TSDB_CODE_SUCCESS) {
×
718
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
719
  }
720
  return code;
×
721
}
722

723
int32_t getColInfoResultForGroupby(void* pVnode, SNodeList* group, STableListInfo* pTableListInfo, uint8_t* digest,
62,016✔
724
                                   SStorageAPI* pAPI, bool initRemainGroups) {
725
  int32_t      code = TSDB_CODE_SUCCESS;
62,016✔
726
  int32_t      lino = 0;
62,016✔
727
  SArray*      pBlockList = NULL;
62,016✔
728
  SSDataBlock* pResBlock = NULL;
62,016✔
729
  void*        keyBuf = NULL;
62,016✔
730
  SArray*      groupData = NULL;
62,016✔
731
  SArray*      pUidTagList = NULL;
62,016✔
732
  SArray*      tableList = NULL;
62,016✔
733

734
  int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);
62,016✔
735
  if (rows == 0) {
62,016!
736
    return TSDB_CODE_SUCCESS;
×
737
  }
738

739
  tagFilterAssist ctx = {0};
62,016✔
740
  ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
62,016✔
741
  if (ctx.colHash == NULL) {
62,041!
742
    code = terrno;
×
743
    goto end;
×
744
  }
745

746
  ctx.index = 0;
62,041✔
747
  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
62,041✔
748
  if (ctx.cInfoList == NULL) {
62,047✔
749
    code = terrno;
15✔
750
    goto end;
×
751
  }
752

753
  SNode* pNode = NULL;
62,032✔
754
  FOREACH(pNode, group) {
128,392!
755
    nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);
66,342✔
756
    if (TSDB_CODE_SUCCESS != ctx.code) {
66,360!
757
      code = ctx.code;
×
758
      goto end;
×
759
    }
760
    REPLACE_NODE(pNode);
66,360✔
761
  }
762

763
  T_MD5_CTX context = {0};
62,050✔
764
  if (tsTagFilterCache) {
62,050!
765
    SNodeListNode* listNode = NULL;
×
766
    code = nodesMakeNode(QUERY_NODE_NODE_LIST, (SNode**)&listNode);
×
767
    if (TSDB_CODE_SUCCESS != code) {
×
768
      goto end;
×
769
    }
770
    listNode->pNodeList = group;
×
771
    code = genTbGroupDigest((SNode*)listNode, digest, &context);
×
772
    QUERY_CHECK_CODE(code, lino, end);
×
773

774
    nodesFree(listNode);
×
775

776
    code = pAPI->metaFn.metaGetCachedTbGroup(pVnode, pTableListInfo->idInfo.suid, context.digest,
×
777
                                             tListLen(context.digest), &tableList);
778
    QUERY_CHECK_CODE(code, lino, end);
×
779

780
    if (tableList) {
×
781
      taosArrayDestroy(pTableListInfo->pTableList);
×
782
      pTableListInfo->pTableList = tableList;
×
783
      qDebug("retrieve tb group list from cache, numOfTables:%d",
×
784
             (int32_t)taosArrayGetSize(pTableListInfo->pTableList));
785
      goto end;
×
786
    }
787
  }
788

789
  pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));
62,050✔
790
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
62,002!
791

792
  for (int32_t i = 0; i < rows; ++i) {
306,017✔
793
    STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
243,947✔
794
    QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);
243,807!
795
    STUidTagInfo info = {.uid = pkeyInfo->uid};
243,807✔
796
    void*        tmp = taosArrayPush(pUidTagList, &info);
244,010✔
797
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
244,010!
798
  }
799

800
  code = pAPI->metaFn.getTableTags(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
62,070✔
801
  if (code != TSDB_CODE_SUCCESS) {
62,051!
802
    goto end;
×
803
  }
804

805
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
62,051✔
806
  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);
62,031✔
807
  if (pResBlock == NULL) {
62,032!
808
    code = terrno;
×
809
    goto end;
×
810
  }
811

812
  //  int64_t st1 = taosGetTimestampUs();
813
  //  qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
814

815
  pBlockList = taosArrayInit(2, POINTER_BYTES);
62,032✔
816
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
62,038!
817

818
  void* tmp = taosArrayPush(pBlockList, &pResBlock);
62,041✔
819
  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
62,041!
820

821
  groupData = taosArrayInit(2, POINTER_BYTES);
62,041✔
822
  QUERY_CHECK_NULL(groupData, code, lino, end, terrno);
62,046!
823

824
  FOREACH(pNode, group) {
128,394!
825
    SScalarParam output = {0};
66,338✔
826

827
    switch (nodeType(pNode)) {
66,338!
828
      case QUERY_NODE_VALUE:
×
829
        break;
×
830
      case QUERY_NODE_COLUMN:
66,338✔
831
      case QUERY_NODE_OPERATOR:
832
      case QUERY_NODE_FUNCTION: {
833
        SExprNode* expNode = (SExprNode*)pNode;
66,338✔
834
        code = createResultData(&expNode->resType, rows, &output);
66,338✔
835
        if (code != TSDB_CODE_SUCCESS) {
66,357!
836
          goto end;
×
837
        }
838
        break;
66,357✔
839
      }
840

841
      default:
×
842
        code = TSDB_CODE_OPS_NOT_SUPPORT;
×
843
        goto end;
×
844
    }
845

846
    if (nodeType(pNode) == QUERY_NODE_COLUMN) {
66,357✔
847
      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;
65,903✔
848
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);
65,903✔
849
      QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);
65,861!
850
      code = colDataAssign(output.columnData, pColInfo, rows, NULL);
65,861✔
851
    } else if (nodeType(pNode) == QUERY_NODE_VALUE) {
454!
852
      continue;
×
853
    } else {
854
      code = scalarCalculate(pNode, pBlockList, &output);
454✔
855
    }
856

857
    if (code != TSDB_CODE_SUCCESS) {
66,344!
858
      releaseColInfoData(output.columnData);
×
859
      goto end;
×
860
    }
861

862
    void* tmp = taosArrayPush(groupData, &output.columnData);
66,347✔
863
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
66,347!
864
  }
865

866
  int32_t keyLen = 0;
62,056✔
867
  SNode*  node;
868
  FOREACH(node, group) {
128,398✔
869
    SExprNode* pExpr = (SExprNode*)node;
66,342✔
870
    keyLen += pExpr->resType.bytes;
66,342✔
871
  }
872

873
  int32_t nullFlagSize = sizeof(int8_t) * LIST_LENGTH(group);
62,056✔
874
  keyLen += nullFlagSize;
62,056✔
875

876
  keyBuf = taosMemoryCalloc(1, keyLen);
62,056✔
877
  if (keyBuf == NULL) {
62,033!
878
    code = terrno;
×
879
    goto end;
×
880
  }
881

882
  if (initRemainGroups) {
62,033✔
883
    pTableListInfo->remainGroups =
14,326✔
884
        taosHashInit(rows, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
14,337✔
885
    if (pTableListInfo->remainGroups == NULL) {
14,326!
886
      code = terrno;
×
887
      goto end;
×
888
    }
889
  }
890

891
  for (int i = 0; i < rows; i++) {
305,924✔
892
    STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
243,844✔
893
    QUERY_CHECK_NULL(info, code, lino, end, terrno);
243,662✔
894

895
    char* isNull = (char*)keyBuf;
243,653✔
896
    char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(group);
243,653!
897
    for (int j = 0; j < taosArrayGetSize(groupData); j++) {
500,239✔
898
      SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);
256,678✔
899

900
      if (colDataIsNull_s(pValue, i)) {
513,202✔
901
        isNull[j] = 1;
506✔
902
      } else {
903
        isNull[j] = 0;
256,095✔
904
        char* data = colDataGetData(pValue, i);
256,095!
905
        if (pValue->info.type == TSDB_DATA_TYPE_JSON) {
256,095✔
906
          if (tTagIsJson(data)) {
264✔
907
            code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
12✔
908
            goto end;
12✔
909
          }
910
          if (tTagIsJsonNull(data)) {
252!
911
            isNull[j] = 1;
×
912
            continue;
×
913
          }
914
          int32_t len = getJsonValueLen(data);
252✔
915
          memcpy(pStart, data, len);
252✔
916
          pStart += len;
252✔
917
        } else if (IS_VAR_DATA_TYPE(pValue->info.type)) {
255,831!
918
          if (varDataTLen(data) > pValue->info.bytes) {
230,986✔
919
            code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
3✔
920
            goto end;
3✔
921
          }
922
          memcpy(pStart, data, varDataTLen(data));
230,983✔
923
          pStart += varDataTLen(data);
230,983✔
924
        } else {
925
          memcpy(pStart, data, pValue->info.bytes);
24,845✔
926
          pStart += pValue->info.bytes;
24,845✔
927
        }
928
      }
929
    }
930

931
    int32_t len = (int32_t)(pStart - (char*)keyBuf);
242,788✔
932
    info->groupId = calcGroupId(keyBuf, len);
242,788✔
933
    if (initRemainGroups) {
243,893✔
934
      // groupId ~ table uid
935
      code = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid),
82,444✔
936
                         sizeof(info->uid));
937
      if (code == TSDB_CODE_DUP_KEY) {
82,453✔
938
        code = TSDB_CODE_SUCCESS;
2,375✔
939
      }
940
      QUERY_CHECK_CODE(code, lino, end);
82,453!
941
    }
942
  }
943

944
  if (tsTagFilterCache) {
62,080!
945
    tableList = taosArrayDup(pTableListInfo->pTableList, NULL);
×
946
    QUERY_CHECK_NULL(tableList, code, lino, end, terrno);
×
947

948
    code = pAPI->metaFn.metaPutTbGroupToCache(pVnode, pTableListInfo->idInfo.suid, context.digest,
×
949
                                              tListLen(context.digest), tableList,
950
                                              taosArrayGetSize(tableList) * sizeof(STableKeyInfo));
×
951
    QUERY_CHECK_CODE(code, lino, end);
×
952
  }
953

954
  //  int64_t st2 = taosGetTimestampUs();
955
  //  qDebug("calculate tag block rows:%d, cost:%ld us", rows, st2-st1);
956

957
end:
62,080✔
958
  taosMemoryFreeClear(keyBuf);
62,095!
959
  taosHashCleanup(ctx.colHash);
62,091✔
960
  taosArrayDestroy(ctx.cInfoList);
62,045✔
961
  blockDataDestroy(pResBlock);
62,054✔
962
  taosArrayDestroy(pBlockList);
62,053✔
963
  taosArrayDestroyEx(pUidTagList, freeItem);
62,052✔
964
  taosArrayDestroyP(groupData, releaseColInfoData);
62,048✔
965
  if (code != TSDB_CODE_SUCCESS) {
62,053✔
966
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
15!
967
  }
968
  return code;
62,039✔
969
}
970

971
static int32_t nameComparFn(const void* p1, const void* p2) {
4,076✔
972
  const char* pName1 = *(const char**)p1;
4,076✔
973
  const char* pName2 = *(const char**)p2;
4,076✔
974

975
  int32_t ret = strcmp(pName1, pName2);
4,076✔
976
  if (ret == 0) {
4,076✔
977
    return 0;
34✔
978
  } else {
979
    return (ret > 0) ? 1 : -1;
4,042✔
980
  }
981
}
982

983
static SArray* getTableNameList(const SNodeListNode* pList) {
854,775✔
984
  int32_t    code = TSDB_CODE_SUCCESS;
854,775✔
985
  int32_t    lino = 0;
854,775✔
986
  int32_t    len = LIST_LENGTH(pList->pNodeList);
854,775!
987
  SListCell* cell = pList->pNodeList->pHead;
854,775✔
988

989
  SArray* pTbList = taosArrayInit(len, POINTER_BYTES);
854,775✔
990
  QUERY_CHECK_NULL(pTbList, code, lino, _end, terrno);
854,784!
991

992
  for (int i = 0; i < pList->pNodeList->length; i++) {
1,712,421✔
993
    SValueNode* valueNode = (SValueNode*)cell->pNode;
857,640✔
994
    if (!IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {
857,640!
995
      terrno = TSDB_CODE_INVALID_PARA;
×
996
      taosArrayDestroy(pTbList);
×
997
      return NULL;
×
998
    }
999

1000
    char* name = varDataVal(valueNode->datum.p);
857,640✔
1001
    void* tmp = taosArrayPush(pTbList, &name);
857,637✔
1002
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
857,637!
1003
    cell = cell->pNext;
857,637✔
1004
  }
1005

1006
  size_t numOfTables = taosArrayGetSize(pTbList);
854,781✔
1007

1008
  // order the name
1009
  taosArraySort(pTbList, nameComparFn);
854,769✔
1010

1011
  // remove the duplicates
1012
  SArray* pNewList = taosArrayInit(taosArrayGetSize(pTbList), sizeof(void*));
854,780✔
1013
  QUERY_CHECK_NULL(pNewList, code, lino, _end, terrno);
854,786!
1014
  void* tmpTbl = taosArrayGet(pTbList, 0);
854,786✔
1015
  QUERY_CHECK_NULL(tmpTbl, code, lino, _end, terrno);
854,765!
1016
  void* tmp = taosArrayPush(pNewList, tmpTbl);
854,771✔
1017
  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
854,771!
1018

1019
  for (int32_t i = 1; i < numOfTables; ++i) {
857,630✔
1020
    char** name = taosArrayGetLast(pNewList);
2,859✔
1021
    char** nameInOldList = taosArrayGet(pTbList, i);
2,860✔
1022
    QUERY_CHECK_NULL(nameInOldList, code, lino, _end, terrno);
2,860!
1023
    if (strcmp(*name, *nameInOldList) == 0) {
2,860✔
1024
      continue;
28✔
1025
    }
1026

1027
    tmp = taosArrayPush(pNewList, nameInOldList);
2,831✔
1028
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
2,831!
1029
  }
1030

1031
_end:
854,771✔
1032
  taosArrayDestroy(pTbList);
854,771✔
1033
  if (code != TSDB_CODE_SUCCESS) {
854,783✔
1034
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
2!
1035
    return NULL;
×
1036
  }
1037
  return pNewList;
854,781✔
1038
}
1039

1040
static int tableUidCompare(const void* a, const void* b) {
×
1041
  uint64_t u1 = *(uint64_t*)a;
×
1042
  uint64_t u2 = *(uint64_t*)b;
×
1043

1044
  if (u1 == u2) {
×
1045
    return 0;
×
1046
  }
1047

1048
  return u1 < u2 ? -1 : 1;
×
1049
}
1050

1051
static int32_t filterTableInfoCompare(const void* a, const void* b) {
101,701✔
1052
  STUidTagInfo* p1 = (STUidTagInfo*)a;
101,701✔
1053
  STUidTagInfo* p2 = (STUidTagInfo*)b;
101,701✔
1054

1055
  if (p1->uid == p2->uid) {
101,701!
1056
    return 0;
×
1057
  }
1058

1059
  return p1->uid < p2->uid ? -1 : 1;
101,701!
1060
}
1061

1062
static FilterCondType checkTagCond(SNode* cond) {
2,461,148✔
1063
  if (nodeType(cond) == QUERY_NODE_OPERATOR) {
2,461,148✔
1064
    return FILTER_NO_LOGIC;
582,615✔
1065
  }
1066
  if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
1,878,533!
1067
    return FILTER_AND;
21,730✔
1068
  }
1069
  return FILTER_OTHER;
1,856,803✔
1070
}
1071

1072
static int32_t optimizeTbnameInCond(void* pVnode, int64_t suid, SArray* list, SNode* cond, SStorageAPI* pAPI) {
2,461,233✔
1073
  int32_t ret = -1;
2,461,233✔
1074
  int32_t ntype = nodeType(cond);
2,461,233✔
1075

1076
  if (ntype == QUERY_NODE_OPERATOR) {
2,461,233✔
1077
    ret = optimizeTbnameInCondImpl(pVnode, list, cond, pAPI, suid);
582,698✔
1078
  }
1079

1080
  if (ntype != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
2,463,513✔
1081
    return ret;
605,977✔
1082
  }
1083

1084
  bool                 hasTbnameCond = false;
1,857,536✔
1085
  SLogicConditionNode* pNode = (SLogicConditionNode*)cond;
1,857,536✔
1086
  SNodeList*           pList = (SNodeList*)pNode->pParameterList;
1,857,536✔
1087

1088
  int32_t len = LIST_LENGTH(pList);
1,857,536✔
1089
  if (len <= 0) {
1,857,536!
1090
    return ret;
×
1091
  }
1092

1093
  SListCell* cell = pList->pHead;
1,857,536✔
1094
  for (int i = 0; i < len; i++) {
6,283,362✔
1095
    if (cell == NULL) break;
4,933,718!
1096
    if (optimizeTbnameInCondImpl(pVnode, list, cell->pNode, pAPI, suid) == 0) {
4,933,718✔
1097
      hasTbnameCond = true;
509,476✔
1098
      break;
509,476✔
1099
    }
1100
    cell = cell->pNext;
4,425,826✔
1101
  }
1102

1103
  taosArraySort(list, filterTableInfoCompare);
1,859,120✔
1104
  taosArrayRemoveDuplicate(list, filterTableInfoCompare, NULL);
1,856,154✔
1105

1106
  if (hasTbnameCond) {
1,856,988✔
1107
    ret = pAPI->metaFn.getTableTagsByUid(pVnode, suid, list);
509,478✔
1108
  }
1109

1110
  return ret;
1,857,141✔
1111
}
1112

1113
// only return uid that does not contained in pExistedUidList
1114
static int32_t optimizeTbnameInCondImpl(void* pVnode, SArray* pExistedUidList, SNode* pTagCond, SStorageAPI* pStoreAPI,
5,514,725✔
1115
                                        uint64_t suid) {
1116
  if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {
5,514,725✔
1117
    return -1;
491,363✔
1118
  }
1119

1120
  SOperatorNode* pNode = (SOperatorNode*)pTagCond;
5,023,362✔
1121
  if (pNode->opType != OP_TYPE_IN) {
5,023,362✔
1122
    return -1;
3,453,432✔
1123
  }
1124

1125
  if ((pNode->pLeft != NULL && nodeType(pNode->pLeft) == QUERY_NODE_COLUMN &&
1,569,930!
1126
       ((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME) &&
1,574,886✔
1127
      (pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {
854,777!
1128
    SNodeListNode* pList = (SNodeListNode*)pNode->pRight;
854,777✔
1129

1130
    int32_t len = LIST_LENGTH(pList->pNodeList);
854,777!
1131
    if (len <= 0) {
854,777!
1132
      return -1;
×
1133
    }
1134

1135
    SArray*   pTbList = getTableNameList(pList);
854,777✔
1136
    int32_t   numOfTables = taosArrayGetSize(pTbList);
854,780✔
1137
    SHashObj* uHash = NULL;
854,776✔
1138

1139
    size_t numOfExisted = taosArrayGetSize(pExistedUidList);  // len > 0 means there already have uids
854,776✔
1140
    if (numOfExisted > 0) {
854,766✔
1141
      uHash = taosHashInit(numOfExisted / 0.7, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
13,288✔
1142
      if (!uHash) {
13,288!
1143
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1144
        return terrno;
×
1145
      }
1146

1147
      for (int i = 0; i < numOfExisted; i++) {
51,439✔
1148
        STUidTagInfo* pTInfo = taosArrayGet(pExistedUidList, i);
38,150✔
1149
        if (!pTInfo) {
38,150!
1150
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1151
          return terrno;
×
1152
        }
1153
        int32_t tempRes = taosHashPut(uHash, &pTInfo->uid, sizeof(uint64_t), &i, sizeof(i));
38,150✔
1154
        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
38,151!
1155
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
1156
          return tempRes;
×
1157
        }
1158
      }
1159
    }
1160

1161
    for (int i = 0; i < numOfTables; i++) {
1,712,319✔
1162
      char* name = taosArrayGetP(pTbList, i);
857,568✔
1163

1164
      uint64_t uid = 0, csuid = 0;
857,562✔
1165
      if (pStoreAPI->metaFn.getTableUidByName(pVnode, name, &uid) == 0) {
857,562✔
1166
        ETableType tbType = TSDB_TABLE_MAX;
854,777✔
1167
        if (pStoreAPI->metaFn.getTableTypeSuidByName(pVnode, name, &tbType, &csuid) == 0 &&
854,777!
1168
            tbType == TSDB_CHILD_TABLE) {
854,781✔
1169
          if (suid != csuid) {
854,751✔
1170
            continue;
20✔
1171
          }
1172
          if (NULL == uHash || taosHashGet(uHash, &uid, sizeof(uid)) == NULL) {
854,731✔
1173
            STUidTagInfo s = {.uid = uid, .name = name, .pTagVal = NULL};
841,448✔
1174
            void*        tmp = taosArrayPush(pExistedUidList, &s);
841,441✔
1175
            if (!tmp) {
841,441!
1176
              return terrno;
×
1177
            }
1178
          }
1179
        } else {
1180
          taosArrayDestroy(pTbList);
30✔
1181
          taosHashCleanup(uHash);
29✔
1182
          return -1;
29✔
1183
        }
1184
      } else {
1185
        //        qWarn("failed to get tableIds from by table name: %s, reason: %s", name, tstrerror(terrno));
1186
        terrno = 0;
2,809✔
1187
      }
1188
    }
1189

1190
    taosHashCleanup(uHash);
854,751✔
1191
    taosArrayDestroy(pTbList);
854,746✔
1192
    return 0;
854,752✔
1193
  }
1194

1195
  return -1;
715,153✔
1196
}
1197

1198
SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, SArray* pUidTagList, void* pVnode,
2,462,806✔
1199
                                        SStorageAPI* pStorageAPI) {
1200
  int32_t      code = TSDB_CODE_SUCCESS;
2,462,806✔
1201
  int32_t      lino = 0;
2,462,806✔
1202
  SSDataBlock* pResBlock = NULL;
2,462,806✔
1203
  code = createDataBlock(&pResBlock);
2,462,806✔
1204
  QUERY_CHECK_CODE(code, lino, _end);
2,464,208!
1205

1206
  for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {
8,407,596✔
1207
    SColumnInfoData colInfo = {0};
5,934,119✔
1208
    void*           tmp = taosArrayGet(pColList, i);
5,934,119✔
1209
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
5,922,773!
1210
    colInfo.info = *(SColumnInfo*)tmp;
5,922,773✔
1211
    code = blockDataAppendColInfo(pResBlock, &colInfo);
5,922,773✔
1212
    QUERY_CHECK_CODE(code, lino, _end);
5,943,388!
1213
  }
1214

1215
  code = blockDataEnsureCapacity(pResBlock, numOfTables);
2,454,485✔
1216
  if (code != TSDB_CODE_SUCCESS) {
2,465,079!
1217
    terrno = code;
×
1218
    blockDataDestroy(pResBlock);
×
1219
    return NULL;
×
1220
  }
1221

1222
  pResBlock->info.rows = numOfTables;
2,465,079✔
1223

1224
  int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);
2,465,079✔
1225

1226
  for (int32_t i = 0; i < numOfTables; i++) {
8,112,264✔
1227
    STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);
5,606,151✔
1228
    QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
5,592,010!
1229

1230
    for (int32_t j = 0; j < numOfCols; j++) {
18,784,302✔
1231
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, j);
13,134,151✔
1232
      QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
13,076,168!
1233

1234
      if (pColInfo->info.colId == -1) {  // tbname
13,080,367✔
1235
        char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
897,345✔
1236
        if (p1->name != NULL) {
897,345✔
1237
          STR_TO_VARSTR(str, p1->name);
841,444✔
1238
        } else {  // name is not retrieved during filter
1239
          code = pStorageAPI->metaFn.getTableNameByUid(pVnode, p1->uid, str);
55,901✔
1240
          QUERY_CHECK_CODE(code, lino, _end);
56,075!
1241
        }
1242

1243
        code = colDataSetVal(pColInfo, i, str, false);
897,519✔
1244
        QUERY_CHECK_CODE(code, lino, _end);
897,484!
1245
#if TAG_FILTER_DEBUG
1246
        qDebug("tagfilter uid:%ld, tbname:%s", *uid, str + 2);
1247
#endif
1248
      } else {
1249
        STagVal tagVal = {0};
12,183,022✔
1250
        tagVal.cid = pColInfo->info.colId;
12,183,022✔
1251
        if (p1->pTagVal == NULL) {
12,183,022!
1252
          colDataSetNULL(pColInfo, i);
×
1253
        } else {
1254
          const char* p = pStorageAPI->metaFn.extractTagVal(p1->pTagVal, pColInfo->info.type, &tagVal);
12,183,022✔
1255

1256
          if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {
12,374,217✔
1257
            colDataSetNULL(pColInfo, i);
3,147✔
1258
          } else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) {
12,371,070✔
1259
            code = colDataSetVal(pColInfo, i, p, false);
4,860✔
1260
            QUERY_CHECK_CODE(code, lino, _end);
4,860!
1261
          } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
15,987,445!
1262
            char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);
3,625,896!
1263
            QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
3,623,301!
1264
            varDataSetLen(tmp, tagVal.nData);
3,623,301✔
1265
            memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);
3,623,301✔
1266
            code = colDataSetVal(pColInfo, i, tmp, false);
3,623,301✔
1267
#if TAG_FILTER_DEBUG
1268
            qDebug("tagfilter varch:%s", tmp + 2);
1269
#endif
1270
            taosMemoryFree(tmp);
3,609,548!
1271
            QUERY_CHECK_CODE(code, lino, _end);
3,621,235!
1272
          } else {
1273
            code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false);
8,740,314✔
1274
            QUERY_CHECK_CODE(code, lino, _end);
8,665,262!
1275
#if TAG_FILTER_DEBUG
1276
            if (pColInfo->info.type == TSDB_DATA_TYPE_INT) {
1277
              qDebug("tagfilter int:%d", *(int*)(&tagVal.i64));
1278
            } else if (pColInfo->info.type == TSDB_DATA_TYPE_DOUBLE) {
1279
              qDebug("tagfilter double:%f", *(double*)(&tagVal.i64));
1280
            }
1281
#endif
1282
          }
1283
        }
1284
      }
1285
    }
1286
  }
1287

1288
_end:
2,506,113✔
1289
  if (code != TSDB_CODE_SUCCESS) {
2,506,113!
1290
    blockDataDestroy(pResBlock);
×
1291
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1292
    terrno = code;
×
1293
    return NULL;
×
1294
  }
1295
  return pResBlock;
2,506,113✔
1296
}
1297

1298
static int32_t doSetQualifiedUid(STableListInfo* pListInfo, SArray* pUidList, const SArray* pUidTagList,
2,386,310✔
1299
                                 bool* pResultList, bool addUid) {
1300
  taosArrayClear(pUidList);
2,386,310✔
1301

1302
  STableKeyInfo info = {.uid = 0, .groupId = 0};
2,386,188✔
1303
  int32_t       numOfTables = taosArrayGetSize(pUidTagList);
2,386,188✔
1304
  for (int32_t i = 0; i < numOfTables; ++i) {
7,740,596✔
1305
    if (pResultList[i]) {
5,352,872✔
1306
      STUidTagInfo* tmpTag = (STUidTagInfo*)taosArrayGet(pUidTagList, i);
4,293,848✔
1307
      if (!tmpTag) {
4,285,196!
1308
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1309
        return terrno;
×
1310
      }
1311
      uint64_t uid = tmpTag->uid;
4,285,196✔
1312
      qDebug("tagfilter get uid:%" PRId64 ", res:%d", uid, pResultList[i]);
4,285,196✔
1313

1314
      info.uid = uid;
4,285,225✔
1315
      void* p = taosArrayPush(pListInfo->pTableList, &info);
4,285,225✔
1316
      if (p == NULL) {
4,295,894!
1317
        return terrno;
×
1318
      }
1319

1320
      if (addUid) {
4,295,894!
1321
        void* tmp = taosArrayPush(pUidList, &uid);
×
1322
        if (tmp == NULL) {
×
1323
          return terrno;
×
1324
        }
1325
      }
1326
    }
1327
  }
1328

1329
  return TSDB_CODE_SUCCESS;
2,387,724✔
1330
}
1331

1332
static int32_t copyExistedUids(SArray* pUidTagList, const SArray* pUidList) {
2,462,634✔
1333
  int32_t code = TSDB_CODE_SUCCESS;
2,462,634✔
1334
  int32_t numOfExisted = taosArrayGetSize(pUidList);
2,462,634✔
1335
  if (numOfExisted == 0) {
2,461,446✔
1336
    return code;
2,380,622✔
1337
  }
1338

1339
  for (int32_t i = 0; i < numOfExisted; ++i) {
258,619✔
1340
    uint64_t* uid = taosArrayGet(pUidList, i);
177,753✔
1341
    if (!uid) {
177,649!
1342
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1343
      return terrno;
×
1344
    }
1345
    STUidTagInfo info = {.uid = *uid};
177,649✔
1346
    void*        tmp = taosArrayPush(pUidTagList, &info);
177,795✔
1347
    if (!tmp) {
177,795!
1348
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1349
      return code;
×
1350
    }
1351
  }
1352
  return code;
80,866✔
1353
}
1354

1355
static int32_t doFilterByTagCond(STableListInfo* pListInfo, SArray* pUidList, SNode* pTagCond, void* pVnode,
8,539,278✔
1356
                                 SIdxFltStatus status, SStorageAPI* pAPI, bool addUid, bool* listAdded) {
1357
  *listAdded = false;
8,539,278✔
1358
  if (pTagCond == NULL) {
8,539,278✔
1359
    return TSDB_CODE_SUCCESS;
6,078,034✔
1360
  }
1361

1362
  terrno = TSDB_CODE_SUCCESS;
2,461,244✔
1363

1364
  int32_t      lino = 0;
2,464,281✔
1365
  int32_t      code = TSDB_CODE_SUCCESS;
2,464,281✔
1366
  SArray*      pBlockList = NULL;
2,464,281✔
1367
  SSDataBlock* pResBlock = NULL;
2,464,281✔
1368
  SScalarParam output = {0};
2,464,281✔
1369
  SArray*      pUidTagList = NULL;
2,464,281✔
1370

1371
  tagFilterAssist ctx = {0};
2,464,281✔
1372
  ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
2,464,281✔
1373
  if (ctx.colHash == NULL) {
2,465,814!
1374
    code = terrno;
×
1375
    QUERY_CHECK_CODE(code, lino, end);
×
1376
  }
1377

1378
  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
2,465,814✔
1379
  if (ctx.cInfoList == NULL) {
2,466,334!
1380
    code = terrno;
×
1381
    QUERY_CHECK_CODE(code, lino, end);
×
1382
  }
1383

1384
  nodesRewriteExprPostOrder(&pTagCond, getColumn, (void*)&ctx);
2,466,334✔
1385
  if (TSDB_CODE_SUCCESS != ctx.code) {
2,463,538!
1386
    terrno = code = ctx.code;
×
1387
    goto end;
×
1388
  }
1389

1390
  SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};
2,463,538✔
1391

1392
  //  int64_t stt = taosGetTimestampUs();
1393
  pUidTagList = taosArrayInit(10, sizeof(STUidTagInfo));
2,463,538✔
1394
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
2,463,457!
1395

1396
  code = copyExistedUids(pUidTagList, pUidList);
2,463,457✔
1397
  QUERY_CHECK_CODE(code, lino, end);
2,461,908!
1398

1399
  FilterCondType condType = checkTagCond(pTagCond);
2,461,908✔
1400

1401
  int32_t filter = optimizeTbnameInCond(pVnode, pListInfo->idInfo.suid, pUidTagList, pTagCond, pAPI);
2,461,765✔
1402
  if (filter == 0) {  // tbname in filter is activated, do nothing and return
2,461,721✔
1403
    taosArrayClear(pUidList);
854,760✔
1404

1405
    int32_t numOfRows = taosArrayGetSize(pUidTagList);
854,755✔
1406
    code = taosArrayEnsureCap(pUidList, numOfRows);
854,746✔
1407
    QUERY_CHECK_CODE(code, lino, end);
854,745!
1408

1409
    for (int32_t i = 0; i < numOfRows; ++i) {
1,734,338✔
1410
      STUidTagInfo* pInfo = taosArrayGet(pUidTagList, i);
879,603✔
1411
      QUERY_CHECK_NULL(pInfo, code, lino, end, terrno);
879,585!
1412
      void* tmp = taosArrayPush(pUidList, &pInfo->uid);
879,585✔
1413
      QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
879,593!
1414
    }
1415
    terrno = 0;
854,735✔
1416
  } else {
1417
    if ((condType == FILTER_NO_LOGIC || condType == FILTER_AND) && status != SFLT_NOT_INDEX) {
1,606,961✔
1418
      code = pAPI->metaFn.getTableTagsByUid(pVnode, pListInfo->idInfo.suid, pUidTagList);
6,677✔
1419
    } else {
1420
      code = pAPI->metaFn.getTableTags(pVnode, pListInfo->idInfo.suid, pUidTagList);
1,600,284✔
1421
    }
1422
    if (code != TSDB_CODE_SUCCESS) {
1,611,676!
1423
      qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), pListInfo->idInfo.suid);
×
1424
      terrno = code;
×
1425
      QUERY_CHECK_CODE(code, lino, end);
×
1426
    }
1427
  }
1428

1429
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
2,466,414✔
1430
  if (numOfTables == 0) {
2,463,549✔
1431
    goto end;
78,306✔
1432
  }
1433

1434
  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);
2,385,243✔
1435
  if (pResBlock == NULL) {
2,385,859!
1436
    code = terrno;
×
1437
    QUERY_CHECK_CODE(code, lino, end);
×
1438
  }
1439

1440
  //  int64_t st1 = taosGetTimestampUs();
1441
  //  qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
1442
  pBlockList = taosArrayInit(2, POINTER_BYTES);
2,385,859✔
1443
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
2,388,047!
1444

1445
  void* tmp = taosArrayPush(pBlockList, &pResBlock);
2,387,932✔
1446
  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
2,387,932!
1447

1448
  code = createResultData(&type, numOfTables, &output);
2,387,932✔
1449
  if (code != TSDB_CODE_SUCCESS) {
2,388,290!
1450
    terrno = code;
×
1451
    QUERY_CHECK_CODE(code, lino, end);
×
1452
  }
1453

1454
  code = scalarCalculate(pTagCond, pBlockList, &output);
2,388,290✔
1455
  if (code != TSDB_CODE_SUCCESS) {
2,387,591✔
1456
    qError("failed to calculate scalar, reason:%s", tstrerror(code));
8!
1457
    terrno = code;
8✔
1458
    QUERY_CHECK_CODE(code, lino, end);
8!
1459
  }
1460

1461
  code = doSetQualifiedUid(pListInfo, pUidList, pUidTagList, (bool*)output.columnData->pData, addUid);
2,387,583✔
1462
  if (code != TSDB_CODE_SUCCESS) {
2,387,362!
1463
    terrno = code;
×
1464
    QUERY_CHECK_CODE(code, lino, end);
×
1465
  }
1466
  *listAdded = true;
2,387,407✔
1467

1468
end:
2,465,721✔
1469
  if (code != TSDB_CODE_SUCCESS) {
2,465,721✔
1470
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
8!
1471
  }
1472
  taosHashCleanup(ctx.colHash);
2,465,721✔
1473
  taosArrayDestroy(ctx.cInfoList);
2,466,483✔
1474
  blockDataDestroy(pResBlock);
2,466,829✔
1475
  taosArrayDestroy(pBlockList);
2,466,726✔
1476
  taosArrayDestroyEx(pUidTagList, freeItem);
2,466,833✔
1477

1478
  colDataDestroy(output.columnData);
2,466,507✔
1479
  taosMemoryFreeClear(output.columnData);
2,466,960!
1480
  return code;
2,466,954✔
1481
}
1482

1483
int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
8,534,271✔
1484
                     STableListInfo* pListInfo, uint8_t* digest, const char* idstr, SStorageAPI* pStorageAPI) {
1485
  int32_t code = TSDB_CODE_SUCCESS;
8,534,271✔
1486
  int32_t lino = 0;
8,534,271✔
1487
  size_t  numOfTables = 0;
8,534,271✔
1488
  bool    listAdded = false;
8,534,271✔
1489

1490
  pListInfo->idInfo.suid = pScanNode->suid;
8,534,271✔
1491
  pListInfo->idInfo.tableType = pScanNode->tableType;
8,534,271✔
1492

1493
  SArray* pUidList = taosArrayInit(8, sizeof(uint64_t));
8,534,271✔
1494
  QUERY_CHECK_NULL(pUidList, code, lino, _error, terrno);
8,544,377✔
1495

1496
  SIdxFltStatus status = SFLT_NOT_INDEX;
8,538,949✔
1497
  if (pScanNode->tableType != TSDB_SUPER_TABLE) {
8,538,949✔
1498
    pListInfo->idInfo.uid = pScanNode->uid;
1,082,437✔
1499
    if (pStorageAPI->metaFn.isTableExisted(pVnode, pScanNode->uid)) {
1,082,437✔
1500
      void* tmp = taosArrayPush(pUidList, &pScanNode->uid);
1,079,732✔
1501
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
1,079,726!
1502
    }
1503
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, false, &listAdded);
1,082,436✔
1504
    QUERY_CHECK_CODE(code, lino, _end);
1,082,422!
1505
  } else {
1506
    T_MD5_CTX context = {0};
7,456,512✔
1507

1508
    if (tsTagFilterCache) {
7,456,512!
1509
      // try to retrieve the result from meta cache
1510
      code = genTagFilterDigest(pTagCond, &context);
×
1511
      QUERY_CHECK_CODE(code, lino, _error);
×
1512

1513
      bool acquired = false;
×
1514
      code = pStorageAPI->metaFn.getCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),
×
1515
                                                    pUidList, &acquired);
1516
      QUERY_CHECK_CODE(code, lino, _error);
×
1517

1518
      if (acquired) {
×
1519
        digest[0] = 1;
×
1520
        memcpy(digest + 1, context.digest, tListLen(context.digest));
×
1521
        qDebug("retrieve table uid list from cache, numOfTables:%d", (int32_t)taosArrayGetSize(pUidList));
×
1522
        goto _end;
×
1523
      }
1524
    }
1525

1526
    if (!pTagCond) {  // no tag filter condition exists, let's fetch all tables of this super table
7,456,512✔
1527
      code = pStorageAPI->metaFn.getChildTableList(pVnode, pScanNode->suid, pUidList);
5,045,267✔
1528
      QUERY_CHECK_CODE(code, lino, _error);
5,055,485!
1529
    } else {
1530
      // failed to find the result in the cache, let try to calculate the results
1531
      if (pTagIndexCond) {
2,411,245✔
1532
        void* pIndex = pStorageAPI->metaFn.getInvertIndex(pVnode);
106,673✔
1533

1534
        SIndexMetaArg metaArg = {.metaEx = pVnode,
213,457✔
1535
                                 .idx = pStorageAPI->metaFn.storeGetIndexInfo(pVnode),
106,763✔
1536
                                 .ivtIdx = pIndex,
1537
                                 .suid = pScanNode->uid};
106,694✔
1538

1539
        status = SFLT_NOT_INDEX;
106,694✔
1540
        code = doFilterTag(pTagIndexCond, &metaArg, pUidList, &status, &pStorageAPI->metaFilter);
106,694✔
1541
        if (code != 0 || status == SFLT_NOT_INDEX) {  // temporarily disable it for performance sake
106,736✔
1542
          qDebug("failed to get tableIds from index, suid:%" PRIu64, pScanNode->uid);
80,997✔
1543
        } else {
1544
          qDebug("succ to get filter result, table num: %d", (int)taosArrayGetSize(pUidList));
25,739✔
1545
        }
1546
      }
1547
    }
1548

1549
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, tsTagFilterCache, &listAdded);
7,466,747✔
1550
    QUERY_CHECK_CODE(code, lino, _end);
7,461,643✔
1551

1552
    // let's add the filter results into meta-cache
1553
    numOfTables = taosArrayGetSize(pUidList);
7,461,635✔
1554

1555
    if (tsTagFilterCache) {
7,453,656✔
1556
      size_t size = numOfTables * sizeof(uint64_t) + sizeof(int32_t);
155✔
1557
      char*  pPayload = taosMemoryMalloc(size);
155!
1558
      QUERY_CHECK_NULL(pPayload, code, lino, _end, terrno);
×
1559

1560
      *(int32_t*)pPayload = numOfTables;
×
1561
      if (numOfTables > 0) {
×
1562
        void* tmp = taosArrayGet(pUidList, 0);
×
1563
        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
×
1564
        memcpy(pPayload + sizeof(int32_t), tmp, numOfTables * sizeof(uint64_t));
×
1565
      }
1566

1567
      code = pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),
×
1568
                                                    pPayload, size, 1);
1569
      QUERY_CHECK_CODE(code, lino, _error);
×
1570

1571
      digest[0] = 1;
×
1572
      memcpy(digest + 1, context.digest, tListLen(context.digest));
×
1573
    }
1574
  }
1575

1576
_end:
8,535,931✔
1577
  if (!listAdded) {
8,535,931✔
1578
    numOfTables = taosArrayGetSize(pUidList);
6,150,256✔
1579
    for (int i = 0; i < numOfTables; i++) {
23,116,218✔
1580
      void* tmp = taosArrayGet(pUidList, i);
16,956,619✔
1581
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
16,919,692!
1582
      STableKeyInfo info = {.uid = *(uint64_t*)tmp, .groupId = 0};
16,919,692✔
1583

1584
      void* p = taosArrayPush(pListInfo->pTableList, &info);
16,919,692✔
1585
      if (p == NULL) {
16,966,224!
1586
        taosArrayDestroy(pUidList);
×
1587
        return terrno;
×
1588
      }
1589

1590
      qTrace("tagfilter get uid:%" PRIu64 ", %s", info.uid, idstr);
16,966,224✔
1591
    }
1592
  }
1593

1594
_error:
8,545,274✔
1595
  taosArrayDestroy(pUidList);
8,545,274✔
1596
  if (code != TSDB_CODE_SUCCESS) {
8,547,693✔
1597
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
8!
1598
  }
1599
  return code;
8,547,633✔
1600
}
1601

1602
int32_t qGetTableList(int64_t suid, void* pVnode, void* node, SArray** tableList, void* pTaskInfo) {
81✔
1603
  int32_t        code = TSDB_CODE_SUCCESS;
81✔
1604
  int32_t        lino = 0;
81✔
1605
  SSubplan*      pSubplan = (SSubplan*)node;
81✔
1606
  SScanPhysiNode pNode = {0};
81✔
1607
  pNode.suid = suid;
81✔
1608
  pNode.uid = suid;
81✔
1609
  pNode.tableType = TSDB_SUPER_TABLE;
81✔
1610

1611
  STableListInfo* pTableListInfo = tableListCreate();
81✔
1612
  QUERY_CHECK_NULL(pTableListInfo, code, lino, _end, terrno);
81!
1613
  uint8_t digest[17] = {0};
81✔
1614
  code = getTableList(pVnode, &pNode, pSubplan ? pSubplan->pTagCond : NULL, pSubplan ? pSubplan->pTagIndexCond : NULL,
81✔
1615
                      pTableListInfo, digest, "qGetTableList", &((SExecTaskInfo*)pTaskInfo)->storageAPI);
1616
  QUERY_CHECK_CODE(code, lino, _end);
81!
1617
  *tableList = pTableListInfo->pTableList;
81✔
1618
  pTableListInfo->pTableList = NULL;
81✔
1619
  tableListDestroy(pTableListInfo);
81✔
1620

1621
_end:
81✔
1622
  if (code != TSDB_CODE_SUCCESS) {
81!
1623
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1624
  }
1625
  return code;
81✔
1626
}
1627

1628
size_t getTableTagsBufLen(const SNodeList* pGroups) {
7,869✔
1629
  size_t keyLen = 0;
7,869✔
1630

1631
  SNode* node;
1632
  FOREACH(node, pGroups) {
32,953!
1633
    SExprNode* pExpr = (SExprNode*)node;
25,084✔
1634
    keyLen += pExpr->resType.bytes;
25,084✔
1635
  }
1636

1637
  keyLen += sizeof(int8_t) * LIST_LENGTH(pGroups);
7,869!
1638
  return keyLen;
7,869✔
1639
}
1640

1641
int32_t getGroupIdFromTagsVal(void* pVnode, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId,
24✔
1642
                              SStorageAPI* pAPI) {
1643
  SMetaReader mr = {0};
24✔
1644

1645
  pAPI->metaReaderFn.initReader(&mr, pVnode, META_READER_LOCK, &pAPI->metaFn);
24✔
1646
  if (pAPI->metaReaderFn.getEntryGetUidCache(&mr, uid) != 0) {  // table not exist
24!
1647
    pAPI->metaReaderFn.clearReader(&mr);
×
1648
    return TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
1649
  }
1650

1651
  SNodeList* groupNew = NULL;
24✔
1652
  int32_t    code = nodesCloneList(pGroupNode, &groupNew);
24✔
1653
  if (TSDB_CODE_SUCCESS != code) {
24!
1654
    pAPI->metaReaderFn.clearReader(&mr);
×
1655
    return code;
×
1656
  }
1657

1658
  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
24✔
1659
  nodesRewriteExprsPostOrder(groupNew, doTranslateTagExpr, &ctx);
24✔
1660
  if (TSDB_CODE_SUCCESS != ctx.code) {
24!
1661
    nodesDestroyList(groupNew);
×
1662
    pAPI->metaReaderFn.clearReader(&mr);
×
1663
    return code;
×
1664
  }
1665
  char* isNull = (char*)keyBuf;
24✔
1666
  char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(pGroupNode);
24!
1667

1668
  SNode*  pNode;
1669
  int32_t index = 0;
24✔
1670
  FOREACH(pNode, groupNew) {
48!
1671
    SNode*  pNew = NULL;
24✔
1672
    int32_t code = scalarCalculateConstants(pNode, &pNew);
24✔
1673
    if (TSDB_CODE_SUCCESS == code) {
24!
1674
      REPLACE_NODE(pNew);
24✔
1675
    } else {
1676
      nodesDestroyList(groupNew);
×
1677
      pAPI->metaReaderFn.clearReader(&mr);
×
1678
      return code;
×
1679
    }
1680

1681
    if (nodeType(pNew) != QUERY_NODE_VALUE) {
24!
1682
      nodesDestroyList(groupNew);
×
1683
      pAPI->metaReaderFn.clearReader(&mr);
×
1684
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
1685
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
1686
    }
1687
    SValueNode* pValue = (SValueNode*)pNew;
24✔
1688

1689
    if (pValue->node.resType.type == TSDB_DATA_TYPE_NULL || pValue->isNull) {
24!
1690
      isNull[index++] = 1;
×
1691
      continue;
×
1692
    } else {
1693
      isNull[index++] = 0;
24✔
1694
      char* data = nodesGetValueFromNode(pValue);
24✔
1695
      if (pValue->node.resType.type == TSDB_DATA_TYPE_JSON) {
24!
1696
        if (tTagIsJson(data)) {
×
1697
          terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
×
1698
          nodesDestroyList(groupNew);
×
1699
          pAPI->metaReaderFn.clearReader(&mr);
×
1700
          return terrno;
×
1701
        }
1702
        int32_t len = getJsonValueLen(data);
×
1703
        memcpy(pStart, data, len);
×
1704
        pStart += len;
×
1705
      } else if (IS_VAR_DATA_TYPE(pValue->node.resType.type)) {
24!
1706
        memcpy(pStart, data, varDataTLen(data));
4✔
1707
        pStart += varDataTLen(data);
4✔
1708
      } else {
1709
        memcpy(pStart, data, pValue->node.resType.bytes);
20✔
1710
        pStart += pValue->node.resType.bytes;
20✔
1711
      }
1712
    }
1713
  }
1714

1715
  int32_t len = (int32_t)(pStart - (char*)keyBuf);
24✔
1716
  *pGroupId = calcGroupId(keyBuf, len);
24✔
1717

1718
  nodesDestroyList(groupNew);
24✔
1719
  pAPI->metaReaderFn.clearReader(&mr);
24✔
1720

1721
  return TSDB_CODE_SUCCESS;
24✔
1722
}
1723

1724
SArray* makeColumnArrayFromList(SNodeList* pNodeList) {
38,356✔
1725
  if (!pNodeList) {
38,356!
1726
    return NULL;
×
1727
  }
1728

1729
  size_t  numOfCols = LIST_LENGTH(pNodeList);
38,356!
1730
  SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
38,356✔
1731
  if (pList == NULL) {
38,362!
1732
    return NULL;
×
1733
  }
1734

1735
  for (int32_t i = 0; i < numOfCols; ++i) {
136,069✔
1736
    SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i);
97,697✔
1737
    if (!pColNode) {
97,638!
1738
      taosArrayDestroy(pList);
×
1739
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
1740
      return NULL;
×
1741
    }
1742

1743
    // todo extract method
1744
    SColumn c = {0};
97,638✔
1745
    c.slotId = pColNode->slotId;
97,638✔
1746
    c.colId = pColNode->colId;
97,638✔
1747
    c.type = pColNode->node.resType.type;
97,638✔
1748
    c.bytes = pColNode->node.resType.bytes;
97,638✔
1749
    c.precision = pColNode->node.resType.precision;
97,638✔
1750
    c.scale = pColNode->node.resType.scale;
97,638✔
1751

1752
    void* tmp = taosArrayPush(pList, &c);
97,707✔
1753
    if (!tmp) {
97,707!
1754
      taosArrayDestroy(pList);
×
1755
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1756
      return NULL;
×
1757
    }
1758
  }
1759

1760
  return pList;
38,372✔
1761
}
1762

1763
int32_t extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
16,251,115✔
1764
                            int32_t type, SColMatchInfo* pMatchInfo) {
1765
  size_t  numOfCols = LIST_LENGTH(pNodeList);
16,251,115!
1766
  int32_t code = TSDB_CODE_SUCCESS;
16,251,115✔
1767
  int32_t lino = 0;
16,251,115✔
1768

1769
  pMatchInfo->matchType = type;
16,251,115✔
1770

1771
  SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchItem));
16,251,115✔
1772
  if (pList == NULL) {
16,251,660✔
1773
    code = terrno;
1,052✔
1774
    return code;
×
1775
  }
1776

1777
  for (int32_t i = 0; i < numOfCols; ++i) {
83,899,548✔
1778
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);
67,627,627✔
1779
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
67,577,015!
1780
    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {
67,585,236✔
1781
      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
67,485,553✔
1782

1783
      SColMatchItem c = {.needOutput = true};
67,485,553✔
1784
      c.colId = pColNode->colId;
67,485,553✔
1785
      c.srcSlotId = pColNode->slotId;
67,485,553✔
1786
      c.dstSlotId = pNode->slotId;
67,485,553✔
1787
      c.isPk = pColNode->isPk;
67,485,553✔
1788
      c.dataType = pColNode->node.resType;
67,485,553✔
1789
      void* tmp = taosArrayPush(pList, &c);
67,549,257✔
1790
      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
67,549,257!
1791
    }
1792
  }
1793

1794
  // set the output flag for each column in SColMatchInfo, according to the
1795
  *numOfOutputCols = 0;
16,271,921✔
1796
  int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);
16,271,921✔
1797
  for (int32_t i = 0; i < num; ++i) {
96,732,798✔
1798
    SSlotDescNode* pNode = (SSlotDescNode*)nodesListGetNode(pOutputNodeList->pSlots, i);
80,421,775✔
1799
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
81,158,271!
1800

1801
    // todo: add reserve flag check
1802
    // it is a column reserved for the arithmetic expression calculation
1803
    if (pNode->slotId >= numOfCols) {
81,232,397✔
1804
      (*numOfOutputCols) += 1;
13,702,521✔
1805
      continue;
13,702,521✔
1806
    }
1807

1808
    SColMatchItem* info = NULL;
67,529,876✔
1809
    for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {
2,147,483,647✔
1810
      info = taosArrayGet(pList, j);
2,147,483,647✔
1811
      QUERY_CHECK_NULL(info, code, lino, _end, terrno);
2,147,483,647✔
1812
      if (info->dstSlotId == pNode->slotId) {
2,147,483,647✔
1813
        break;
66,292,321✔
1814
      }
1815
    }
1816

1817
    if (pNode->output) {
66,758,356✔
1818
      (*numOfOutputCols) += 1;
66,682,073✔
1819
    } else if (info != NULL) {
76,283✔
1820
      // select distinct tbname from stb where tbname='abc';
1821
      info->needOutput = false;
76,282✔
1822
    }
1823
  }
1824

1825
  pMatchInfo->pList = pList;
16,311,023✔
1826

1827
_end:
16,311,023✔
1828
  if (code != TSDB_CODE_SUCCESS) {
16,311,023!
1829
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1830
  }
1831
  return code;
16,226,998✔
1832
}
1833

1834
static SResSchema createResSchema(int32_t type, int32_t bytes, int32_t slotId, int32_t scale, int32_t precision,
53,930,944✔
1835
                                  const char* name) {
1836
  SResSchema s = {0};
53,930,944✔
1837
  s.scale = scale;
53,930,944✔
1838
  s.type = type;
53,930,944✔
1839
  s.bytes = bytes;
53,930,944✔
1840
  s.slotId = slotId;
53,930,944✔
1841
  s.precision = precision;
53,930,944✔
1842
  tstrncpy(s.name, name, tListLen(s.name));
53,930,944✔
1843

1844
  return s;
53,930,944✔
1845
}
1846

1847
static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType, EColumnType colType) {
51,947,934✔
1848
  SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn));
51,947,934!
1849
  if (pCol == NULL) {
51,926,543!
1850
    return NULL;
×
1851
  }
1852

1853
  pCol->slotId = slotId;
51,926,543✔
1854
  pCol->colId = colId;
51,926,543✔
1855
  pCol->bytes = pType->bytes;
51,926,543✔
1856
  pCol->type = pType->type;
51,926,543✔
1857
  pCol->scale = pType->scale;
51,926,543✔
1858
  pCol->precision = pType->precision;
51,926,543✔
1859
  pCol->dataBlockId = blockId;
51,926,543✔
1860
  pCol->colType = colType;
51,926,543✔
1861
  return pCol;
51,926,543✔
1862
}
1863

1864
int32_t createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) {
53,878,306✔
1865
  int32_t code = TSDB_CODE_SUCCESS;
53,878,306✔
1866
  int32_t lino = 0;
53,878,306✔
1867
  pExp->base.numOfParams = 0;
53,878,306✔
1868
  pExp->base.pParam = NULL;
53,878,306✔
1869
  pExp->pExpr = taosMemoryCalloc(1, sizeof(tExprNode));
53,878,306!
1870
  QUERY_CHECK_NULL(pExp->pExpr, code, lino, _end, terrno);
53,942,919!
1871

1872
  pExp->pExpr->_function.num = 1;
53,942,919✔
1873
  pExp->pExpr->_function.functionId = -1;
53,942,919✔
1874

1875
  int32_t type = nodeType(pNode);
53,942,919✔
1876
  // it is a project query, or group by column
1877
  if (type == QUERY_NODE_COLUMN) {
53,942,919✔
1878
    pExp->pExpr->nodeType = QUERY_NODE_COLUMN;
34,194,372✔
1879
    SColumnNode* pColNode = (SColumnNode*)pNode;
34,194,372✔
1880

1881
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
34,194,372✔
1882
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
34,191,678!
1883

1884
    pExp->base.numOfParams = 1;
34,191,678✔
1885

1886
    SDataType* pType = &pColNode->node.resType;
34,191,678✔
1887
    pExp->base.resSchema =
1888
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pColNode->colName);
34,191,678✔
1889

1890
    pExp->base.pParam[0].pCol =
68,380,645✔
1891
        createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType, pColNode->colType);
34,207,787✔
1892
    QUERY_CHECK_NULL(pExp->base.pParam[0].pCol, code, lino, _end, terrno);
34,172,858!
1893

1894
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN;
34,172,858✔
1895
  } else if (type == QUERY_NODE_VALUE) {
19,748,547✔
1896
    pExp->pExpr->nodeType = QUERY_NODE_VALUE;
419,574✔
1897
    SValueNode* pValNode = (SValueNode*)pNode;
419,574✔
1898

1899
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
419,574!
1900
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
419,663!
1901

1902
    pExp->base.numOfParams = 1;
419,663✔
1903

1904
    SDataType* pType = &pValNode->node.resType;
419,663✔
1905
    pExp->base.resSchema =
1906
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pValNode->node.aliasName);
419,663✔
1907
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;
419,612✔
1908
    code = nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param);
419,612✔
1909
    QUERY_CHECK_CODE(code, lino, _end);
401,024!
1910
  } else if (type == QUERY_NODE_FUNCTION) {
19,328,973✔
1911
    pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;
19,004,620✔
1912
    SFunctionNode* pFuncNode = (SFunctionNode*)pNode;
19,004,620✔
1913

1914
    SDataType* pType = &pFuncNode->node.resType;
19,004,620✔
1915
    pExp->base.resSchema =
1916
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pFuncNode->node.aliasName);
19,004,620✔
1917
    tExprNode* pExprNode = pExp->pExpr;
19,009,094✔
1918

1919
    pExprNode->_function.functionId = pFuncNode->funcId;
19,009,094✔
1920
    pExprNode->_function.pFunctNode = pFuncNode;
19,009,094✔
1921
    pExprNode->_function.functionType = pFuncNode->funcType;
19,009,094✔
1922

1923
    tstrncpy(pExprNode->_function.functionName, pFuncNode->functionName, tListLen(pExprNode->_function.functionName));
19,009,094✔
1924

1925
#if 1
1926
    // todo refactor: add the parameter for tbname function
1927
    const char* name = "tbname";
19,009,094✔
1928
    int32_t     len = strlen(name);
19,009,094✔
1929

1930
    if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) &&
19,009,094✔
1931
        pExprNode->_function.functionName[len] == 0) {
1,134,193!
1932
      pFuncNode->pParameterList = NULL;
1,134,264✔
1933
      int32_t     code = nodesMakeList(&pFuncNode->pParameterList);
1,134,264✔
1934
      SValueNode* res = NULL;
1,134,173✔
1935
      if (TSDB_CODE_SUCCESS == code) {
1,134,173!
1936
        code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
1,134,196✔
1937
      }
1938
      QUERY_CHECK_CODE(code, lino, _end);
1,134,143!
1939
      res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
1,134,143✔
1940
      code = nodesListAppend(pFuncNode->pParameterList, (SNode*)res);
1,134,143✔
1941
      if (code != TSDB_CODE_SUCCESS) {
1,134,287!
1942
        nodesDestroyNode((SNode*)res);
×
1943
        res = NULL;
×
1944
      }
1945
      QUERY_CHECK_CODE(code, lino, _end);
1,134,291!
1946
    }
1947
#endif
1948

1949
    int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);
19,009,121✔
1950

1951
    pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam));
19,009,121!
1952
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
19,009,836!
1953
    pExp->base.numOfParams = numOfParam;
19,009,836✔
1954

1955
    for (int32_t j = 0; j < numOfParam && TSDB_CODE_SUCCESS == code; ++j) {
40,833,415!
1956
      SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j);
21,824,265✔
1957
      QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
21,823,839!
1958
      if (p1->type == QUERY_NODE_COLUMN) {
21,826,772✔
1959
        SColumnNode* pcn = (SColumnNode*)p1;
17,755,328✔
1960

1961
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN;
17,755,328✔
1962
        pExp->base.pParam[j].pCol =
35,511,627✔
1963
            createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType, pcn->colType);
17,755,328✔
1964
        QUERY_CHECK_NULL(pExp->base.pParam[j].pCol, code, lino, _end, terrno);
17,756,299✔
1965
      } else if (p1->type == QUERY_NODE_VALUE) {
4,071,444✔
1966
        SValueNode* pvn = (SValueNode*)p1;
3,488,089✔
1967
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE;
3,488,089✔
1968
        code = nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param);
3,488,089✔
1969
        QUERY_CHECK_CODE(code, lino, _end);
3,487,438!
1970
      }
1971
    }
1972
    pExp->pExpr->_function.bindExprID = ((SExprNode*)pNode)->bindExprID;
19,009,150✔
1973
  } else if (type == QUERY_NODE_OPERATOR) {
324,353✔
1974
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
321,266✔
1975
    SOperatorNode* pOpNode = (SOperatorNode*)pNode;
321,266✔
1976

1977
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
321,266!
1978
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
321,281!
1979
    pExp->base.numOfParams = 1;
321,281✔
1980

1981
    SDataType* pType = &pOpNode->node.resType;
321,281✔
1982
    pExp->base.resSchema =
1983
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pOpNode->node.aliasName);
321,281✔
1984
    pExp->pExpr->_optrRoot.pRootNode = pNode;
321,261✔
1985
  } else if (type == QUERY_NODE_CASE_WHEN) {
3,087!
1986
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
4,876✔
1987
    SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;
4,876✔
1988

1989
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
4,876!
1990
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
4,878!
1991
    pExp->base.numOfParams = 1;
4,878✔
1992

1993
    SDataType* pType = &pCaseNode->node.resType;
4,878✔
1994
    pExp->base.resSchema =
1995
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCaseNode->node.aliasName);
4,878✔
1996
    pExp->pExpr->_optrRoot.pRootNode = pNode;
4,876✔
1997
  } else if (type == QUERY_NODE_LOGIC_CONDITION) {
×
1998
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
10✔
1999
    SLogicConditionNode* pCond = (SLogicConditionNode*)pNode;
10✔
2000
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
10!
2001
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
10!
2002
    pExp->base.numOfParams = 1;
10✔
2003
    SDataType* pType = &pCond->node.resType;
10✔
2004
    pExp->base.resSchema =
2005
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCond->node.aliasName);
10✔
2006
    pExp->pExpr->_optrRoot.pRootNode = pNode;
10✔
2007
  } else {
2008
    code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2009
    QUERY_CHECK_CODE(code, lino, _end);
×
2010
  }
2011
  pExp->pExpr->relatedTo = ((SExprNode*)pNode)->relatedTo;
53,907,380✔
2012
_end:
53,907,380✔
2013
  if (code != TSDB_CODE_SUCCESS) {
53,907,380!
2014
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2015
  }
2016
  return code;
53,904,385✔
2017
}
2018

2019
int32_t createExprFromTargetNode(SExprInfo* pExp, STargetNode* pTargetNode) {
53,845,959✔
2020
  return createExprFromOneNode(pExp, pTargetNode->pExpr, pTargetNode->slotId);
53,845,959✔
2021
}
2022

2023
SExprInfo* createExpr(SNodeList* pNodeList, int32_t* numOfExprs) {
1,744✔
2024
  *numOfExprs = LIST_LENGTH(pNodeList);
1,744!
2025
  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
1,744!
2026
  if (!pExprs) {
1,744!
2027
    return NULL;
×
2028
  }
2029

2030
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
16,979✔
2031
    SExprInfo* pExp = &pExprs[i];
15,234✔
2032
    int32_t    code = createExprFromOneNode(pExp, nodesListGetNode(pNodeList, i), i + UD_TAG_COLUMN_INDEX);
15,234✔
2033
    if (code != TSDB_CODE_SUCCESS) {
15,235!
2034
      taosMemoryFreeClear(pExprs);
×
2035
      terrno = code;
×
2036
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2037
      return NULL;
×
2038
    }
2039
  }
2040

2041
  return pExprs;
1,745✔
2042
}
2043

2044
int32_t createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, SExprInfo** pExprInfo, int32_t* numOfExprs) {
20,747,478✔
2045
  QRY_PARAM_CHECK(pExprInfo);
20,747,478!
2046

2047
  int32_t code = 0;
20,747,478✔
2048
  int32_t numOfFuncs = LIST_LENGTH(pNodeList);
20,747,478✔
2049
  int32_t numOfGroupKeys = 0;
20,747,478✔
2050
  if (pGroupKeys != NULL) {
20,747,478✔
2051
    numOfGroupKeys = LIST_LENGTH(pGroupKeys);
581,140✔
2052
  }
2053

2054
  *numOfExprs = numOfFuncs + numOfGroupKeys;
20,747,478✔
2055
  if (*numOfExprs == 0) {
20,747,478✔
2056
    return code;
2,949,997✔
2057
  }
2058

2059
  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
17,797,481!
2060
  if (pExprs == NULL) {
17,801,224!
2061
    return terrno;
×
2062
  }
2063

2064
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
70,924,136✔
2065
    STargetNode* pTargetNode = NULL;
53,118,671✔
2066
    if (i < numOfFuncs) {
53,118,671✔
2067
      pTargetNode = (STargetNode*)nodesListGetNode(pNodeList, i);
52,285,794✔
2068
    } else {
2069
      pTargetNode = (STargetNode*)nodesListGetNode(pGroupKeys, i - numOfFuncs);
832,877✔
2070
    }
2071
    if (!pTargetNode) {
53,090,017!
2072
      destroyExprInfo(pExprs, *numOfExprs);
×
2073
      taosMemoryFreeClear(pExprs);
×
2074
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2075
      return terrno;
×
2076
    }
2077

2078
    SExprInfo* pExp = &pExprs[i];
53,090,017✔
2079
    code = createExprFromTargetNode(pExp, pTargetNode);
53,090,017✔
2080
    if (code != TSDB_CODE_SUCCESS) {
53,124,020✔
2081
      destroyExprInfo(pExprs, *numOfExprs);
1,108✔
2082
      taosMemoryFreeClear(pExprs);
×
2083
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2084
      return code;
×
2085
    }
2086
  }
2087

2088
  *pExprInfo = pExprs;
17,805,465✔
2089
  return code;
17,805,465✔
2090
}
2091

2092
static void deleteSubsidiareCtx(void* pData) {
×
2093
  SSubsidiaryResInfo* pCtx = (SSubsidiaryResInfo*)pData;
×
2094
  if (pCtx->pCtx) {
×
2095
    taosMemoryFreeClear(pCtx->pCtx);
×
2096
  }
2097
}
×
2098

2099
// set the output buffer for the selectivity + tag query
2100
static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
20,116,334✔
2101
  int32_t num = 0;
20,116,334✔
2102
  int32_t code = TSDB_CODE_SUCCESS;
20,116,334✔
2103
  int32_t lino = 0;
20,116,334✔
2104

2105
  SArray* pValCtxArray = NULL;
20,116,334✔
2106
  for (int32_t i = numOfOutput - 1; i > 0; --i) {  // select Func is at the end of the list
55,181,647✔
2107
    int32_t funcIdx = pCtx[i].pExpr->pExpr->_function.bindExprID;
35,065,314✔
2108
    if (funcIdx > 0) {
35,065,314✔
2109
      if (pValCtxArray == NULL) {
3,551✔
2110
        // the end of the list is the select function of biggest index
2111
        pValCtxArray = taosArrayInit_s(sizeof(SSubsidiaryResInfo*), funcIdx);
2,512✔
2112
        if (pValCtxArray == NULL) {
2,511!
2113
          return terrno;
×
2114
        }
2115
      }
2116
      if (funcIdx > pValCtxArray->size) {
3,550!
2117
        qError("funcIdx:%d is out of range", funcIdx);
×
2118
        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2119
        return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2120
      }
2121
      SSubsidiaryResInfo* pSubsidiary = &pCtx[i].subsidiaries;
3,550✔
2122
      pSubsidiary->pCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
3,550!
2123
      if (pSubsidiary->pCtx == NULL) {
3,552!
2124
        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2125
        return terrno;
×
2126
      }
2127
      pSubsidiary->num = 0;
3,552✔
2128
      taosArraySet(pValCtxArray, funcIdx - 1, &pSubsidiary);
3,552✔
2129
    }
2130
  }
2131

2132
  SqlFunctionCtx*  p = NULL;
20,116,333✔
2133
  SqlFunctionCtx** pValCtx = NULL;
20,116,333✔
2134
  if (pValCtxArray == NULL) {
20,116,333✔
2135
    pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
20,111,326!
2136
    if (pValCtx == NULL) {
20,118,521!
2137
      QUERY_CHECK_CODE(terrno, lino, _end);
×
2138
    }
2139
  }
2140

2141
  for (int32_t i = 0; i < numOfOutput; ++i) {
73,677,649✔
2142
    const char* pName = pCtx[i].pExpr->pExpr->_function.functionName;
53,548,544✔
2143
    if ((strcmp(pName, "_select_value") == 0)) {
53,548,544✔
2144
      if (pValCtxArray == NULL) {
2,262,975✔
2145
        pValCtx[num++] = &pCtx[i];
2,258,557✔
2146
      } else {
2147
        int32_t bindFuncIndex = pCtx[i].pExpr->pExpr->relatedTo;  // start from index 1;
4,418✔
2148
        if (bindFuncIndex > 0) {  // 0 is default index related to the select function
4,418!
2149
          bindFuncIndex -= 1;
4,855✔
2150
        }
2151
        SSubsidiaryResInfo** pSubsidiary = taosArrayGet(pValCtxArray, bindFuncIndex);
4,418✔
2152
        if(pSubsidiary == NULL) {
4,978!
2153
          QUERY_CHECK_CODE(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, lino, _end);
×
2154
        }
2155
        (*pSubsidiary)->pCtx[(*pSubsidiary)->num] = &pCtx[i];
4,978✔
2156
        (*pSubsidiary)->num++;
4,978✔
2157
      }
2158
    } else if (fmIsSelectFunc(pCtx[i].functionId)) {
51,285,569✔
2159
       if (pValCtxArray == NULL) {
4,932,424✔
2160
        p = &pCtx[i];
4,927,630✔
2161
       }
2162
    }
2163
  }
2164

2165
  if (p != NULL) {
20,129,105✔
2166
    p->subsidiaries.pCtx = pValCtx;
4,247,594✔
2167
    p->subsidiaries.num = num;
4,247,594✔
2168
  } else {
2169
    taosMemoryFreeClear(pValCtx);
15,881,511!
2170
  }
2171

2172
_end:
13,571✔
2173
  if (code != TSDB_CODE_SUCCESS) {
20,130,660!
2174
    taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2175
    taosMemoryFreeClear(pValCtx);
×
2176
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2177
  } else {
2178
    taosArrayDestroy(pValCtxArray);
20,130,660✔
2179
  }
2180
  return code;
20,103,975✔
2181
}
2182

2183
SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowEntryInfoOffset,
20,113,142✔
2184
                                     SFunctionStateStore* pStore) {
2185
  int32_t         code = TSDB_CODE_SUCCESS;
20,113,142✔
2186
  int32_t         lino = 0;
20,113,142✔
2187
  SqlFunctionCtx* pFuncCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));
20,113,142!
2188
  if (pFuncCtx == NULL) {
20,111,492!
2189
    return NULL;
×
2190
  }
2191

2192
  *rowEntryInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t));
20,111,492!
2193
  if (*rowEntryInfoOffset == 0) {
20,124,253!
2194
    taosMemoryFreeClear(pFuncCtx);
×
2195
    return NULL;
×
2196
  }
2197

2198
  for (int32_t i = 0; i < numOfOutput; ++i) {
73,687,406✔
2199
    SExprInfo* pExpr = &pExprInfo[i];
53,558,890✔
2200

2201
    SExprBasicInfo* pFunct = &pExpr->base;
53,558,890✔
2202
    SqlFunctionCtx* pCtx = &pFuncCtx[i];
53,558,890✔
2203

2204
    pCtx->functionId = -1;
53,558,890✔
2205
    pCtx->pExpr = pExpr;
53,558,890✔
2206

2207
    if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {
53,558,890✔
2208
      SFuncExecEnv env = {0};
19,004,856✔
2209
      pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;
19,004,856✔
2210
      pCtx->isPseudoFunc = fmIsWindowPseudoColumnFunc(pCtx->functionId);
19,004,856✔
2211
      pCtx->isNotNullFunc = fmIsNotNullOutputFunc(pCtx->functionId);
18,998,097✔
2212

2213
      bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);
19,001,423✔
2214
      if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId)) {
32,052,851✔
2215
        if (!isUdaf) {
13,055,802✔
2216
          code = fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
13,055,770✔
2217
          QUERY_CHECK_CODE(code, lino, _end);
13,052,031!
2218
        } else {
2219
          char* udfName = pExpr->pExpr->_function.pFunctNode->functionName;
32✔
2220
          pCtx->udfName = taosStrdup(udfName);
32!
2221
          QUERY_CHECK_NULL(pCtx->udfName, code, lino, _end, terrno);
32!
2222

2223
          code = fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet);
32✔
2224
          QUERY_CHECK_CODE(code, lino, _end);
32!
2225
        }
2226
        bool tmp = pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
13,052,063✔
2227
        if (!tmp) {
13,051,209!
2228
          code = terrno;
×
2229
          QUERY_CHECK_CODE(code, lino, _end);
×
2230
        }
2231
      } else {
2232
        code = fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);
5,934,615✔
2233
        if (code != TSDB_CODE_SUCCESS && isUdaf) {
5,931,620!
2234
          code = TSDB_CODE_SUCCESS;
261✔
2235
        }
2236
        QUERY_CHECK_CODE(code, lino, _end);
5,931,620!
2237

2238
        if (pCtx->sfp.getEnv != NULL) {
5,931,620✔
2239
          bool tmp = pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
3,503,860✔
2240
          if (!tmp) {
3,501,307✔
2241
            code = terrno;
3,745✔
2242
            QUERY_CHECK_CODE(code, lino, _end);
×
2243
          }
2244
        }
2245
      }
2246
      pCtx->resDataInfo.interBufSize = env.calcMemSize;
18,979,961✔
2247
    } else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR ||
34,554,034✔
2248
               pExpr->pExpr->nodeType == QUERY_NODE_VALUE) {
419,644!
2249
      // for simple column, the result buffer needs to hold at least one element.
2250
      pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes;
34,556,682✔
2251
    }
2252

2253
    pCtx->input.numOfInputCols = pFunct->numOfParams;
53,533,995✔
2254
    pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
53,533,995!
2255
    QUERY_CHECK_NULL(pCtx->input.pData, code, lino, _end, terrno);
53,555,354!
2256
    pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
53,555,354!
2257
    QUERY_CHECK_NULL(pCtx->input.pColumnDataAgg, code, lino, _end, terrno);
53,563,153!
2258

2259
    pCtx->pTsOutput = NULL;
53,563,153✔
2260
    pCtx->resDataInfo.bytes = pFunct->resSchema.bytes;
53,563,153✔
2261
    pCtx->resDataInfo.type = pFunct->resSchema.type;
53,563,153✔
2262
    pCtx->order = TSDB_ORDER_ASC;
53,563,153✔
2263
    pCtx->start.key = INT64_MIN;
53,563,153✔
2264
    pCtx->end.key = INT64_MIN;
53,563,153✔
2265
    pCtx->numOfParams = pExpr->base.numOfParams;
53,563,153✔
2266
    pCtx->param = pFunct->pParam;
53,563,153✔
2267
    pCtx->saveHandle.currentPage = -1;
53,563,153✔
2268
    pCtx->pStore = pStore;
53,563,153✔
2269
    pCtx->hasWindowOrGroup = false;
53,563,153✔
2270
    pCtx->needCleanup = false;
53,563,153✔
2271
  }
2272

2273
  for (int32_t i = 1; i < numOfOutput; ++i) {
55,202,267✔
2274
    (*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) +
35,073,751✔
2275
                                         pFuncCtx[i - 1].resDataInfo.interBufSize);
35,073,751✔
2276
  }
2277

2278
  code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);
20,128,516✔
2279
  QUERY_CHECK_CODE(code, lino, _end);
20,106,420!
2280

2281
_end:
20,106,420✔
2282
  if (code != TSDB_CODE_SUCCESS) {
20,106,420!
2283
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2284
    for (int32_t i = 0; i < numOfOutput; ++i) {
×
2285
      taosMemoryFree(pFuncCtx[i].input.pData);
×
2286
      taosMemoryFree(pFuncCtx[i].input.pColumnDataAgg);
×
2287
    }
2288
    taosMemoryFreeClear(*rowEntryInfoOffset);
×
2289
    taosMemoryFreeClear(pFuncCtx);
×
2290

2291
    terrno = code;
×
2292
    return NULL;
×
2293
  }
2294
  return pFuncCtx;
20,106,420✔
2295
}
2296

2297
// NOTE: sources columns are more than the destination SSDatablock columns.
2298
// doFilter in table scan needs every column even its output is false
2299
int32_t relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols, bool outputEveryColumn) {
2,275,278✔
2300
  int32_t code = TSDB_CODE_SUCCESS;
2,275,278✔
2301
  size_t  numOfSrcCols = taosArrayGetSize(pCols);
2,275,278✔
2302

2303
  int32_t i = 0, j = 0;
2,274,658✔
2304
  while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) {
4,979,492✔
2305
    SColumnInfoData* p = taosArrayGet(pCols, i);
2,703,256✔
2306
    if (!p) {
2,701,697!
2307
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2308
      return terrno;
×
2309
    }
2310
    SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);
2,701,697✔
2311
    if (!pmInfo) {
2,701,014!
2312
      return terrno;
×
2313
    }
2314

2315
    if (p->info.colId == pmInfo->colId) {
2,701,537✔
2316
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->dstSlotId);
2,520,757✔
2317
      if (!pDst) {
2,519,733!
2318
        return terrno;
×
2319
      }
2320
      code = colDataAssign(pDst, p, pBlock->info.rows, &pBlock->info);
2,519,733✔
2321
      if (code != TSDB_CODE_SUCCESS) {
2,523,880!
2322
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2323
        return code;
×
2324
      }
2325
      i++;
2,524,051✔
2326
      j++;
2,524,051✔
2327
    } else if (p->info.colId < pmInfo->colId) {
180,780!
2328
      i++;
180,783✔
2329
    } else {
2330
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
2331
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2332
    }
2333
  }
2334
  return code;
2,275,208✔
2335
}
2336

2337
SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
4,895,107✔
2338
  SInterval interval = {
4,895,107✔
2339
      .interval = pTableScanNode->interval,
4,895,107✔
2340
      .sliding = pTableScanNode->sliding,
4,895,107✔
2341
      .intervalUnit = pTableScanNode->intervalUnit,
4,895,107✔
2342
      .slidingUnit = pTableScanNode->slidingUnit,
4,895,107✔
2343
      .offset = pTableScanNode->offset,
4,895,107✔
2344
      .precision = pTableScanNode->scan.node.pOutputDataBlockDesc->precision,
4,895,107✔
2345
      .timeRange = pTableScanNode->scanRange,
2346
  };
2347
  calcIntervalAutoOffset(&interval);
4,895,107✔
2348

2349
  return interval;
4,895,772✔
2350
}
2351

2352
SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {
1,110,558✔
2353
  SColumn c = {0};
1,110,558✔
2354

2355
  c.slotId = pColNode->slotId;
1,110,558✔
2356
  c.colId = pColNode->colId;
1,110,558✔
2357
  c.type = pColNode->node.resType.type;
1,110,558✔
2358
  c.bytes = pColNode->node.resType.bytes;
1,110,558✔
2359
  c.scale = pColNode->node.resType.scale;
1,110,558✔
2360
  c.precision = pColNode->node.resType.precision;
1,110,558✔
2361
  return c;
1,110,558✔
2362
}
2363

2364
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode,
8,549,074✔
2365
                               const SReadHandle* readHandle) {
2366
  pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
8,549,074✔
2367
  pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);
8,549,074!
2368

2369
  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
8,549,074!
2370
  if (!pCond->colList) {
8,561,689!
2371
    return terrno;
×
2372
  }
2373
  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
8,561,689!
2374
  if (pCond->pSlotList == NULL) {
8,556,957!
2375
    taosMemoryFreeClear(pCond->colList);
×
2376
    return terrno;
×
2377
  }
2378

2379
  // TODO: get it from stable scan node
2380
  pCond->twindows = pTableScanNode->scanRange;
8,556,957✔
2381
  pCond->suid = pTableScanNode->scan.suid;
8,556,957✔
2382
  pCond->type = TIMEWINDOW_RANGE_CONTAINED;
8,556,957✔
2383
  pCond->startVersion = -1;
8,556,957✔
2384
  pCond->endVersion = -1;
8,556,957✔
2385
  pCond->skipRollup = readHandle->skipRollup;
8,556,957✔
2386

2387
  // allowed read stt file optimization mode
2388
  pCond->notLoadData = (pTableScanNode->dataRequired == FUNC_DATA_REQUIRED_NOT_LOAD) &&
17,403,212✔
2389
                       (pTableScanNode->scan.node.pConditions == NULL) && (pTableScanNode->interval == 0);
8,556,957✔
2390

2391
  int32_t j = 0;
8,556,957✔
2392
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
53,740,328✔
2393
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pTableScanNode->scan.pScanCols, i);
45,188,368✔
2394
    if (!pNode) {
45,182,532!
2395
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2396
      return terrno;
×
2397
    }
2398
    SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
45,183,371✔
2399
    if (pColNode->colType == COLUMN_TYPE_TAG) {
45,183,371!
2400
      continue;
×
2401
    }
2402

2403
    pCond->colList[j].type = pColNode->node.resType.type;
45,183,371✔
2404
    pCond->colList[j].bytes = pColNode->node.resType.bytes;
45,183,371✔
2405
    pCond->colList[j].colId = pColNode->colId;
45,183,371✔
2406
    pCond->colList[j].pk = pColNode->isPk;
45,183,371✔
2407

2408
    pCond->pSlotList[j] = pNode->slotId;
45,183,371✔
2409
    j += 1;
45,183,371✔
2410
  }
2411

2412
  pCond->numOfCols = j;
8,551,960✔
2413
  return TSDB_CODE_SUCCESS;
8,551,960✔
2414
}
2415

2416
int32_t initQueryTableDataCondWithColArray(SQueryTableDataCond* pCond, SQueryTableDataCond* pOrgCond,
228✔
2417
                                           const SReadHandle* readHandle, SArray* colArray) {
2418
  int32_t code = TSDB_CODE_SUCCESS;
228✔
2419
  int32_t lino = 0;
228✔
2420

2421
  pCond->order = TSDB_ORDER_ASC;
228✔
2422
  pCond->numOfCols = (int32_t)taosArrayGetSize(colArray);
228✔
2423

2424
  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
228!
2425
  QUERY_CHECK_NULL(pCond->colList, code, lino, _return, terrno);
228!
2426

2427
  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
228!
2428
  QUERY_CHECK_NULL(pCond->pSlotList, code, lino, _return, terrno);
228!
2429

2430
  pCond->twindows = pOrgCond->twindows;
228✔
2431
  pCond->type = pOrgCond->type;
228✔
2432
  pCond->startVersion = -1;
228✔
2433
  pCond->endVersion = -1;
228✔
2434
  pCond->skipRollup = true;
228✔
2435
  pCond->notLoadData = false;
228✔
2436

2437
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
1,380✔
2438
    SColIdPair* pColPair = taosArrayGet(colArray, i);
1,152✔
2439
    QUERY_CHECK_NULL(pColPair, code, lino, _return, terrno);
1,152!
2440

2441
    bool find = false;
1,152✔
2442
    for (int32_t j = 0; j < pOrgCond->numOfCols; ++j) {
8,592!
2443
      if (pOrgCond->colList[j].colId == pColPair->vtbColId) {
8,592✔
2444
        pCond->colList[i].type = pOrgCond->colList[j].type;
1,152✔
2445
        pCond->colList[i].bytes = pOrgCond->colList[j].bytes;
1,152✔
2446
        pCond->colList[i].colId = pColPair->orgColId;
1,152✔
2447
        pCond->colList[i].pk = pOrgCond->colList[j].pk;
1,152✔
2448
        pCond->pSlotList[i] = i;
1,152✔
2449
        find = true;
1,152✔
2450
        break;
1,152✔
2451
      }
2452
    }
2453
    QUERY_CHECK_CONDITION(find, code, lino, _return, TSDB_CODE_NOT_FOUND);
1,152!
2454
  }
2455

2456
  return code;
228✔
2457
_return:
×
2458
  qError("%s failed at line %d since %s", __func__, lino, tstrerror(terrno));
×
2459
  taosMemoryFreeClear(pCond->colList);
×
2460
  taosMemoryFreeClear(pCond->pSlotList);
×
2461
  return code;
×
2462
}
2463

2464
void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {
20,806,192✔
2465
  taosMemoryFreeClear(pCond->colList);
20,806,192!
2466
  taosMemoryFreeClear(pCond->pSlotList);
20,806,688!
2467
}
20,806,584✔
2468

2469
int32_t convertFillType(int32_t mode) {
1,084,206✔
2470
  int32_t type = TSDB_FILL_NONE;
1,084,206✔
2471
  switch (mode) {
1,084,206!
2472
    case FILL_MODE_PREV:
34,058✔
2473
      type = TSDB_FILL_PREV;
34,058✔
2474
      break;
34,058✔
2475
    case FILL_MODE_NONE:
×
2476
      type = TSDB_FILL_NONE;
×
2477
      break;
×
2478
    case FILL_MODE_NULL:
50,131✔
2479
      type = TSDB_FILL_NULL;
50,131✔
2480
      break;
50,131✔
2481
    case FILL_MODE_NULL_F:
58,367✔
2482
      type = TSDB_FILL_NULL_F;
58,367✔
2483
      break;
58,367✔
2484
    case FILL_MODE_NEXT:
30,965✔
2485
      type = TSDB_FILL_NEXT;
30,965✔
2486
      break;
30,965✔
2487
    case FILL_MODE_VALUE:
290,133✔
2488
      type = TSDB_FILL_SET_VALUE;
290,133✔
2489
      break;
290,133✔
2490
    case FILL_MODE_VALUE_F:
580,381✔
2491
      type = TSDB_FILL_SET_VALUE_F;
580,381✔
2492
      break;
580,381✔
2493
    case FILL_MODE_LINEAR:
30,162✔
2494
      type = TSDB_FILL_LINEAR;
30,162✔
2495
      break;
30,162✔
2496
    case FILL_MODE_NEAR:
10,030✔
2497
      type = TSDB_FILL_NEAR;
10,030✔
2498
      break;
10,030✔
2499
    default:
×
2500
      type = TSDB_FILL_NONE;
×
2501
  }
2502

2503
  return type;
1,084,206✔
2504
}
2505

2506
void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery) {
3,660,379✔
2507
  if (ascQuery) {
3,660,379✔
2508
    *w = getAlignQueryTimeWindow(pInterval, ts);
3,167,300✔
2509
  } else {
2510
    // the start position of the first time window in the endpoint that spreads beyond the queried last timestamp
2511
    *w = getAlignQueryTimeWindow(pInterval, ts);
493,079✔
2512

2513
    int64_t key = w->skey;
493,123✔
2514
    while (key < ts) {  // moving towards end
1,629,494✔
2515
      key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);
1,597,939✔
2516
      if (key > ts) {
1,597,938✔
2517
        break;
461,567✔
2518
      }
2519

2520
      w->skey = key;
1,136,371✔
2521
    }
2522
    w->ekey = taosTimeAdd(w->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
493,122✔
2523
  }
2524
}
3,660,362✔
2525

2526
static STimeWindow doCalculateTimeWindow(int64_t ts, SInterval* pInterval) {
1,405,898✔
2527
  STimeWindow w = {0};
1,405,898✔
2528

2529
  w.skey = taosTimeTruncate(ts, pInterval);
1,405,898✔
2530
  w.ekey = taosTimeGetIntervalEnd(w.skey, pInterval);
1,405,959✔
2531
  return w;
1,405,860✔
2532
}
2533

2534
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {
708,782✔
2535
  STimeWindow win = *pWindow;
708,782✔
2536
  STimeWindow save = win;
708,782✔
2537
  while (win.skey <= ts && win.ekey >= ts) {
3,074,402✔
2538
    save = win;
2,365,614✔
2539
    // get previous time window
2540
    getNextTimeWindow(pInterval, &win, order == TSDB_ORDER_ASC ? TSDB_ORDER_DESC : TSDB_ORDER_ASC);
2,365,614✔
2541
  }
2542

2543
  return save;
708,788✔
2544
}
2545

2546
// get the correct time window according to the handled timestamp
2547
// todo refactor
2548
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
5,924,751✔
2549
                                int32_t order) {
2550
  STimeWindow w = {0};
5,924,751✔
2551
  if (pResultRowInfo->cur.pageId == -1) {  // the first window, from the previous stored value
5,924,751✔
2552
    getInitialStartTimeWindow(pInterval, ts, &w, (order == TSDB_ORDER_ASC));
3,660,410✔
2553
    return w;
3,660,356✔
2554
  }
2555

2556
  SResultRow* pRow = getResultRowByPos(pBuf, &pResultRowInfo->cur, false);
2,264,341✔
2557
  if (pRow) {
2,264,370!
2558
    TAOS_SET_OBJ_ALIGNED(&w, pRow->win);
2,264,373✔
2559
  }
2560

2561
  // in case of typical time window, we can calculate time window directly.
2562
  if (w.skey > ts || w.ekey < ts) {
2,264,370✔
2563
    w = doCalculateTimeWindow(ts, pInterval);
1,405,895✔
2564
  }
2565

2566
  if (pInterval->interval != pInterval->sliding) {
2,264,345✔
2567
    // it is an sliding window query, in which sliding value is not equalled to
2568
    // interval value, and we need to find the first qualified time window.
2569
    w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
708,782✔
2570
  }
2571

2572
  return w;
2,264,335✔
2573
}
2574

2575
TSKEY getNextTimeWindowStart(const SInterval* pInterval, TSKEY start, int32_t order) {
355,281,881✔
2576
  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
355,281,881✔
2577
  TSKEY   nextStart = taosTimeAdd(start, -1 * pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
355,281,881✔
2578
  nextStart = taosTimeAdd(nextStart, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision, NULL);
355,147,893✔
2579
  nextStart = taosTimeAdd(nextStart, pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
355,191,718✔
2580
  return nextStart;
355,061,169✔
2581
}
2582

2583
void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t order) {
353,639,244✔
2584
  tw->skey = getNextTimeWindowStart(pInterval, tw->skey, order);
353,639,244✔
2585
  tw->ekey = taosTimeAdd(tw->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
353,405,063✔
2586
}
353,646,890✔
2587

2588
bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {
28,459,769✔
2589
  return (pLimitInfo->limit.limit != -1 || pLimitInfo->limit.offset != -1 || pLimitInfo->slimit.limit != -1 ||
56,895,203!
2590
          pLimitInfo->slimit.offset != -1);
28,435,434✔
2591
}
2592

2593
bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo) {
×
2594
  return (pLimitInfo->slimit.limit != -1 || pLimitInfo->slimit.offset != -1);
×
2595
}
2596

2597
void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo) {
24,880,610✔
2598
  SLimit limit = {.limit = getLimit(pLimit), .offset = getOffset(pLimit)};
24,880,610✔
2599
  SLimit slimit = {.limit = getLimit(pSLimit), .offset = getOffset(pSLimit)};
24,887,619✔
2600

2601
  pLimitInfo->limit = limit;
24,898,057✔
2602
  pLimitInfo->slimit = slimit;
24,898,057✔
2603
  pLimitInfo->remainOffset = limit.offset;
24,898,057✔
2604
  pLimitInfo->remainGroupOffset = slimit.offset;
24,898,057✔
2605
}
24,898,057✔
2606

2607
void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo) {
4,105,530✔
2608
  pLimitInfo->numOfOutputRows = 0;
4,105,530✔
2609
  pLimitInfo->remainOffset = pLimitInfo->limit.offset;
4,105,530✔
2610
}
4,105,530✔
2611

2612
int32_t tableListGetSize(const STableListInfo* pTableList, int32_t* pRes) {
31,446,078✔
2613
  if (taosArrayGetSize(pTableList->pTableList) != taosHashGetSize(pTableList->map)) {
31,446,078!
2614
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
2615
    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2616
  }
2617
  (*pRes) = taosArrayGetSize(pTableList->pTableList);
31,442,358✔
2618
  return TSDB_CODE_SUCCESS;
31,429,329✔
2619
}
2620

2621
uint64_t tableListGetSuid(const STableListInfo* pTableList) { return pTableList->idInfo.suid; }
155,982✔
2622

2623
STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index) {
13,029,869✔
2624
  if (taosArrayGetSize(pTableList->pTableList) == 0) {
13,029,869✔
2625
    return NULL;
1,108✔
2626
  }
2627

2628
  return taosArrayGet(pTableList->pTableList, index);
13,021,600✔
2629
}
2630

2631
int32_t tableListFind(const STableListInfo* pTableList, uint64_t uid, int32_t startIndex) {
595✔
2632
  int32_t numOfTables = taosArrayGetSize(pTableList->pTableList);
595✔
2633
  if (startIndex >= numOfTables) {
595!
2634
    return -1;
×
2635
  }
2636

2637
  for (int32_t i = startIndex; i < numOfTables; ++i) {
4,151!
2638
    STableKeyInfo* p = taosArrayGet(pTableList->pTableList, i);
4,151✔
2639
    if (!p) {
4,151✔
2640
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
1!
2641
      return -1;
×
2642
    }
2643
    if (p->uid == uid) {
4,150✔
2644
      return i;
594✔
2645
    }
2646
  }
2647
  return -1;
×
2648
}
2649

2650
void tableListGetSourceTableInfo(const STableListInfo* pTableList, uint64_t* psuid, uint64_t* uid, int32_t* type) {
23,683✔
2651
  *psuid = pTableList->idInfo.suid;
23,683✔
2652
  *uid = pTableList->idInfo.uid;
23,683✔
2653
  *type = pTableList->idInfo.tableType;
23,683✔
2654
}
23,683✔
2655

2656
uint64_t tableListGetTableGroupId(const STableListInfo* pTableList, uint64_t tableUid) {
27,663,279✔
2657
  int32_t* slot = taosHashGet(pTableList->map, &tableUid, sizeof(tableUid));
27,663,279✔
2658
  if (slot == NULL) {
27,669,532!
2659
    return -1;
×
2660
  }
2661

2662
  STableKeyInfo* pKeyInfo = taosArrayGet(pTableList->pTableList, *slot);
27,669,532✔
2663

2664
  return pKeyInfo->groupId;
27,664,560✔
2665
}
2666

2667
// TODO handle the group offset info, fix it, the rule of group output will be broken by this function
2668
int32_t tableListAddTableInfo(STableListInfo* pTableList, uint64_t uid, uint64_t gid) {
17,672✔
2669
  int32_t code = TSDB_CODE_SUCCESS;
17,672✔
2670
  int32_t lino = 0;
17,672✔
2671
  if (pTableList->map == NULL) {
17,672!
2672
    pTableList->map = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
×
2673
    QUERY_CHECK_NULL(pTableList->map, code, lino, _end, terrno);
×
2674
  }
2675

2676
  STableKeyInfo keyInfo = {.uid = uid, .groupId = gid};
17,672✔
2677
  void*         p = taosHashGet(pTableList->map, &uid, sizeof(uid));
17,672✔
2678
  if (p != NULL) {
17,737✔
2679
    qInfo("table:%" PRId64 " already in tableIdList, ignore it", uid);
3!
2680
    goto _end;
3✔
2681
  }
2682

2683
  void* tmp = taosArrayPush(pTableList->pTableList, &keyInfo);
17,734✔
2684
  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
17,724!
2685

2686
  int32_t slot = (int32_t)taosArrayGetSize(pTableList->pTableList) - 1;
17,724✔
2687
  code = taosHashPut(pTableList->map, &uid, sizeof(uid), &slot, sizeof(slot));
17,695✔
2688
  if (code != TSDB_CODE_SUCCESS) {
17,752!
2689
    // we have checked the existence of uid in hash map above
2690
    QUERY_CHECK_CONDITION((code != TSDB_CODE_DUP_KEY), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
×
2691
    taosArrayPopTailBatch(pTableList->pTableList, 1);  // let's pop the last element in the array list
×
2692
  }
2693

2694
_end:
17,752✔
2695
  if (code != TSDB_CODE_SUCCESS) {
17,749!
2696
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2697
  } else {
2698
    qDebug("uid:%" PRIu64 ", groupId:%" PRIu64 " added into table list, slot:%d, total:%d", uid, gid, slot, slot + 1);
17,749✔
2699
  }
2700

2701
  return code;
17,748✔
2702
}
2703

2704
int32_t tableListGetGroupList(const STableListInfo* pTableList, int32_t ordinalGroupIndex, STableKeyInfo** pKeyInfo,
4,863,811✔
2705
                              int32_t* size) {
2706
  int32_t totalGroups = tableListGetOutputGroups(pTableList);
4,863,811✔
2707
  int32_t numOfTables = 0;
4,871,076✔
2708
  int32_t code = tableListGetSize(pTableList, &numOfTables);
4,871,076✔
2709
  if (code != TSDB_CODE_SUCCESS) {
4,860,239!
2710
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2711
    return code;
×
2712
  }
2713

2714
  if (ordinalGroupIndex < 0 || ordinalGroupIndex >= totalGroups) {
4,860,239!
2715
    return TSDB_CODE_INVALID_PARA;
×
2716
  }
2717

2718
  // here handle two special cases:
2719
  // 1. only one group exists, and 2. one table exists for each group.
2720
  if (totalGroups == 1) {
4,860,239✔
2721
    *size = numOfTables;
4,859,137✔
2722
    *pKeyInfo = (*size == 0) ? NULL : taosArrayGet(pTableList->pTableList, 0);
4,859,137✔
2723
    return TSDB_CODE_SUCCESS;
4,856,692✔
2724
  } else if (totalGroups == numOfTables) {
1,102!
2725
    *size = 1;
1,365✔
2726
    *pKeyInfo = taosArrayGet(pTableList->pTableList, ordinalGroupIndex);
1,365✔
2727
    return TSDB_CODE_SUCCESS;
1,365✔
2728
  }
2729

UNCOV
2730
  int32_t offset = pTableList->groupOffset[ordinalGroupIndex];
×
UNCOV
2731
  if (ordinalGroupIndex < totalGroups - 1) {
×
2732
    *size = pTableList->groupOffset[ordinalGroupIndex + 1] - offset;
543✔
2733
  } else {
2734
    *size = numOfTables - offset;
×
2735
  }
2736

UNCOV
2737
  *pKeyInfo = taosArrayGet(pTableList->pTableList, offset);
×
2738
  return TSDB_CODE_SUCCESS;
680✔
2739
}
2740

2741
int32_t tableListGetOutputGroups(const STableListInfo* pTableList) { return pTableList->numOfOuputGroups; }
14,557,633✔
2742

2743
bool oneTableForEachGroup(const STableListInfo* pTableList) { return pTableList->oneTableForEachGroup; }
20,909✔
2744

2745
STableListInfo* tableListCreate() {
8,629,040✔
2746
  STableListInfo* pListInfo = taosMemoryCalloc(1, sizeof(STableListInfo));
8,629,040!
2747
  if (pListInfo == NULL) {
8,643,509!
2748
    return NULL;
×
2749
  }
2750

2751
  pListInfo->remainGroups = NULL;
8,643,509✔
2752
  pListInfo->pTableList = taosArrayInit(4, sizeof(STableKeyInfo));
8,643,509✔
2753
  if (pListInfo->pTableList == NULL) {
8,642,826!
2754
    goto _error;
×
2755
  }
2756

2757
  pListInfo->map = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
8,642,826✔
2758
  if (pListInfo->map == NULL) {
8,643,918!
2759
    goto _error;
×
2760
  }
2761

2762
  pListInfo->numOfOuputGroups = 1;
8,643,918✔
2763
  return pListInfo;
8,643,918✔
2764

2765
_error:
×
2766
  tableListDestroy(pListInfo);
×
2767
  return NULL;
×
2768
}
2769

2770
void tableListDestroy(STableListInfo* pTableListInfo) {
8,646,576✔
2771
  if (pTableListInfo == NULL) {
8,646,576✔
2772
    return;
243✔
2773
  }
2774

2775
  taosArrayDestroy(pTableListInfo->pTableList);
8,646,333✔
2776
  taosMemoryFreeClear(pTableListInfo->groupOffset);
8,647,428!
2777

2778
  taosHashCleanup(pTableListInfo->map);
8,647,428✔
2779
  taosHashCleanup(pTableListInfo->remainGroups);
8,646,797✔
2780
  pTableListInfo->pTableList = NULL;
8,647,045✔
2781
  pTableListInfo->map = NULL;
8,647,045✔
2782
  taosMemoryFree(pTableListInfo);
8,647,045!
2783
}
2784

2785
void tableListClear(STableListInfo* pTableListInfo) {
2,512✔
2786
  if (pTableListInfo == NULL) {
2,512!
2787
    return;
×
2788
  }
2789

2790
  taosArrayClear(pTableListInfo->pTableList);
2,512✔
2791
  taosHashClear(pTableListInfo->map);
2,512✔
2792
  taosHashClear(pTableListInfo->remainGroups);
2,514✔
2793
  taosMemoryFree(pTableListInfo->groupOffset);
2,514!
2794
  pTableListInfo->numOfOuputGroups = 1;
2,514✔
2795
  pTableListInfo->oneTableForEachGroup = false;
2,514✔
2796
}
2797

2798
static int32_t orderbyGroupIdComparFn(const void* p1, const void* p2) {
108,270✔
2799
  STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;
108,270✔
2800
  STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;
108,270✔
2801

2802
  if (pInfo1->groupId == pInfo2->groupId) {
108,270✔
2803
    return 0;
181✔
2804
  } else {
2805
    return pInfo1->groupId < pInfo2->groupId ? -1 : 1;
108,089✔
2806
  }
2807
}
2808

2809
static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
1,054✔
2810
  taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
1,054✔
2811
  int32_t size = taosArrayGetSize(pTableListInfo->pTableList);
1,056✔
2812

2813
  SArray* pList = taosArrayInit(4, sizeof(int32_t));
1,057✔
2814
  if (!pList) {
1,056!
2815
    return terrno;
×
2816
  }
2817

2818
  STableKeyInfo* pInfo = taosArrayGet(pTableListInfo->pTableList, 0);
1,056✔
2819
  if (!pInfo) {
1,056!
2820
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2821
    return terrno;
×
2822
  }
2823
  uint64_t gid = pInfo->groupId;
1,056✔
2824

2825
  int32_t start = 0;
1,056✔
2826
  void*   tmp = taosArrayPush(pList, &start);
1,056✔
2827
  if (!tmp) {
1,056!
2828
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2829
    return terrno;
×
2830
  }
2831

2832
  for (int32_t i = 1; i < size; ++i) {
3,351✔
2833
    pInfo = taosArrayGet(pTableListInfo->pTableList, i);
2,293✔
2834
    if (!pInfo) {
2,293!
2835
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2836
      return terrno;
×
2837
    }
2838
    if (pInfo->groupId != gid) {
2,294✔
2839
      tmp = taosArrayPush(pList, &i);
2,114✔
2840
      if (!tmp) {
2,114!
2841
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2842
        return terrno;
×
2843
      }
2844
      gid = pInfo->groupId;
2,114✔
2845
    }
2846
  }
2847

2848
  pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);
1,058✔
2849
  pTableListInfo->groupOffset = taosMemoryMalloc(sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
1,055!
2850
  if (pTableListInfo->groupOffset == NULL) {
1,056!
2851
    taosArrayDestroy(pList);
×
2852
    return terrno;
×
2853
  }
2854

2855
  memcpy(pTableListInfo->groupOffset, taosArrayGet(pList, 0), sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
1,056✔
2856
  taosArrayDestroy(pList);
1,056✔
2857
  return TSDB_CODE_SUCCESS;
1,057✔
2858
}
2859

2860
int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle* pHandle, SScanPhysiNode* pScanNode,
8,127,600✔
2861
                                    SNodeList* group, bool groupSort, uint8_t* digest, SStorageAPI* pAPI) {
2862
  int32_t code = TSDB_CODE_SUCCESS;
8,127,600✔
2863

2864
  bool   groupByTbname = groupbyTbname(group);
8,127,600✔
2865
  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
8,128,634✔
2866
  if (!numOfTables) {
8,123,902!
2867
    return code;
×
2868
  }
2869
  if (group == NULL || groupByTbname) {
8,123,902✔
2870
    if (tsCountAlwaysReturnValue && QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode) &&
8,061,875✔
2871
        ((STableScanPhysiNode*)pScanNode)->needCountEmptyTable) {
4,516,920✔
2872
      pTableListInfo->remainGroups =
316,735✔
2873
          taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
316,756✔
2874
      if (pTableListInfo->remainGroups == NULL) {
316,735!
2875
        return terrno;
×
2876
      }
2877

2878
      for (int i = 0; i < numOfTables; i++) {
950,771✔
2879
        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
633,897✔
2880
        if (!info) {
633,859✔
2881
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
12!
2882
          return terrno;
12✔
2883
        }
2884
        info->groupId = groupByTbname ? info->uid : 0;
633,847✔
2885

2886
        int32_t tempRes = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId),
633,847✔
2887
                                      &(info->uid), sizeof(info->uid));
633,847✔
2888
        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
634,036!
2889
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
2890
          return tempRes;
×
2891
        }
2892
      }
2893
    } else {
2894
      for (int32_t i = 0; i < numOfTables; i++) {
28,105,132✔
2895
        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
20,363,686✔
2896
        if (!info) {
20,359,595!
2897
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2898
          return terrno;
×
2899
        }
2900
        info->groupId = groupByTbname ? info->uid : 0;
20,360,013✔
2901
      }
2902
    }
2903

2904
    pTableListInfo->oneTableForEachGroup = groupByTbname;
8,058,320✔
2905
    if (numOfTables == 1 && pTableListInfo->idInfo.tableType == TSDB_CHILD_TABLE) {
8,058,320✔
2906
      pTableListInfo->oneTableForEachGroup = true;
677,083✔
2907
    }
2908

2909
    if (groupSort && groupByTbname) {
8,058,320✔
2910
      taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
39,015✔
2911
      pTableListInfo->numOfOuputGroups = numOfTables;
39,037✔
2912
    } else if (groupByTbname && pScanNode->groupOrderScan) {
8,019,305✔
2913
      pTableListInfo->numOfOuputGroups = numOfTables;
169✔
2914
    } else {
2915
      pTableListInfo->numOfOuputGroups = 1;
8,019,136✔
2916
    }
2917
  } else {
2918
    bool initRemainGroups = false;
62,027✔
2919
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {
62,027✔
2920
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;
61,300✔
2921
      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&
61,300✔
2922
          !(groupSort || pScanNode->groupOrderScan)) {
14,407!
2923
        initRemainGroups = true;
14,342✔
2924
      }
2925
    }
2926

2927
    code = getColInfoResultForGroupby(pHandle->vnode, group, pTableListInfo, digest, pAPI, initRemainGroups);
62,027✔
2928
    if (code != TSDB_CODE_SUCCESS) {
62,036✔
2929
      return code;
15✔
2930
    }
2931

2932
    if (pScanNode->groupOrderScan) pTableListInfo->numOfOuputGroups = taosArrayGetSize(pTableListInfo->pTableList);
62,021✔
2933

2934
    if (groupSort || pScanNode->groupOrderScan) {
62,021✔
2935
      code = sortTableGroup(pTableListInfo);
1,054✔
2936
    }
2937
  }
2938

2939
  // add all table entry in the hash map
2940
  size_t size = taosArrayGetSize(pTableListInfo->pTableList);
8,120,366✔
2941
  for (int32_t i = 0; i < size; ++i) {
29,405,106✔
2942
    STableKeyInfo* p = taosArrayGet(pTableListInfo->pTableList, i);
21,251,508✔
2943
    if (!p) {
21,196,274!
2944
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2945
      return terrno;
×
2946
    }
2947
    int32_t tempRes = taosHashPut(pTableListInfo->map, &p->uid, sizeof(uint64_t), &i, sizeof(int32_t));
21,196,274✔
2948
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
21,284,380!
2949
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
2950
      return tempRes;
×
2951
    }
2952
  }
2953

2954
  return code;
8,153,598✔
2955
}
2956

2957
int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
8,536,140✔
2958
                                STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
2959
                                SExecTaskInfo* pTaskInfo) {
2960
  int64_t     st = taosGetTimestampUs();
8,545,657✔
2961
  const char* idStr = GET_TASKID(pTaskInfo);
8,545,657✔
2962

2963
  if (pHandle == NULL) {
8,545,657!
2964
    qError("invalid handle, in creating operator tree, %s", idStr);
×
2965
    return TSDB_CODE_INVALID_PARA;
×
2966
  }
2967

2968
  uint8_t digest[17] = {0};
8,545,657✔
2969
  int32_t code = getTableList(pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, digest, idStr,
8,545,657✔
2970
                              &pTaskInfo->storageAPI);
2971
  if (code != TSDB_CODE_SUCCESS) {
8,545,550✔
2972
    qError("failed to getTableList, code:%s", tstrerror(code));
8!
2973
    return code;
8✔
2974
  }
2975

2976
  int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
8,545,542✔
2977

2978
  int64_t st1 = taosGetTimestampUs();
8,543,632✔
2979
  pTaskInfo->cost.extractListTime = (st1 - st) / 1000.0;
8,543,632✔
2980
  qDebug("extract queried table list completed, %d tables, elapsed time:%.2f ms %s", numOfTables,
8,543,632✔
2981
         pTaskInfo->cost.extractListTime, idStr);
2982

2983
  if (numOfTables == 0) {
8,544,406✔
2984
    qDebug("no table qualified for query, %s" PRIx64, idStr);
414,791✔
2985
    return TSDB_CODE_SUCCESS;
414,768✔
2986
  }
2987

2988
  code = buildGroupIdMapForAllTables(pTableListInfo, pHandle, pScanNode, pGroupTags, groupSort, digest,
8,129,615✔
2989
                                     &pTaskInfo->storageAPI);
2990
  if (code != TSDB_CODE_SUCCESS) {
8,135,075✔
2991
    return code;
15✔
2992
  }
2993

2994
  pTaskInfo->cost.groupIdMapTime = (taosGetTimestampUs() - st1) / 1000.0;
8,134,508✔
2995
  qDebug("generate group id map completed, elapsed time:%.2f ms %s", pTaskInfo->cost.groupIdMapTime, idStr);
8,134,508✔
2996

2997
  return TSDB_CODE_SUCCESS;
8,134,782✔
2998
}
2999

3000
char* getStreamOpName(uint16_t opType) {
1,463,893✔
3001
  switch (opType) {
1,463,893!
3002
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
184,301✔
3003
      return "stream scan";
184,301✔
3004
    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
41,633✔
3005
      return "project";
41,633✔
3006
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:
637,613✔
3007
      return "interval single";
637,613✔
3008
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL:
15,107✔
3009
      return "interval final";
15,107✔
3010
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
37,593✔
3011
      return "interval semi";
37,593✔
3012
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL:
331✔
3013
      return "interval mid";
331✔
3014
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
8,262✔
3015
      return "stream fill";
8,262✔
3016
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:
11,804✔
3017
      return "session single";
11,804✔
3018
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION:
1,709✔
3019
      return "session semi";
1,709✔
3020
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION:
1,428✔
3021
      return "session final";
1,428✔
3022
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:
9,737✔
3023
      return "state single";
9,737✔
3024
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
479,788✔
3025
      return "stream partitionby";
479,788✔
3026
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT:
1,267✔
3027
      return "stream event";
1,267✔
3028
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT:
16,621✔
3029
      return "stream count";
16,621✔
3030
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC:
9,922✔
3031
      return "stream interp";
9,922✔
3032
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL:
5,440✔
3033
      return "interval continue";
5,440✔
3034
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_INTERVAL:
×
3035
      return "interval continue semi";
×
3036
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_INTERVAL:
×
3037
      return "interval continue final";
×
3038
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SESSION:
×
3039
      return "session continue";
×
3040
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_SESSION:
×
3041
      return "session continue semi";
×
3042
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_SESSION:
×
3043
      return "session continue final";
×
3044
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_STATE:
×
3045
      return "state continue";
×
3046
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_EVENT:
×
3047
      return "event continue";
×
3048
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_COUNT:
×
3049
      return "count continue";
×
3050
  }
3051
  return "error name";
1,337✔
3052
}
3053

3054
void printDataBlock(SSDataBlock* pBlock, const char* flag, const char* taskIdStr) {
646,809✔
3055
  if (!pBlock) {
646,809!
3056
    qDebug("%s===stream===%s: Block is Null", taskIdStr, flag);
×
3057
    return;
×
3058
  } else if (pBlock->info.rows == 0) {
646,809✔
3059
    qDebug("%s===stream===%s: Block is Empty. block type %d", taskIdStr, flag, pBlock->info.type);
29,967✔
3060
    return;
29,969✔
3061
  }
3062
  if (qDebugFlag & DEBUG_DEBUG) {
616,842✔
3063
    char*   pBuf = NULL;
43,294✔
3064
    int32_t code = dumpBlockData(pBlock, flag, &pBuf, taskIdStr);
43,294✔
3065
    if (code == 0) {
43,293!
3066
      qDebug("%s", pBuf);
43,293!
3067
      taosMemoryFree(pBuf);
43,296!
3068
    }
3069
  }
3070
}
3071

3072
void printSpecDataBlock(SSDataBlock* pBlock, const char* flag, const char* opStr, const char* taskIdStr) {
760,693✔
3073
  if (!pBlock) {
760,693!
3074
    qDebug("%s===stream===%s %s: Block is Null", taskIdStr, flag, opStr);
×
3075
    return;
×
3076
  } else if (pBlock->info.rows == 0) {
760,693✔
3077
    qDebug("%s===stream===%s %s: Block is Empty. block type %d.skey:%" PRId64 ",ekey:%" PRId64 ",version%" PRId64,
15,733✔
3078
           taskIdStr, flag, opStr, pBlock->info.type, pBlock->info.window.skey, pBlock->info.window.ekey,
3079
           pBlock->info.version);
3080
    return;
15,733✔
3081
  }
3082
  if (qDebugFlag & DEBUG_DEBUG) {
744,960✔
3083
    char* pBuf = NULL;
49,908✔
3084
    char  flagBuf[64];
3085
    snprintf(flagBuf, sizeof(flagBuf), "%s %s", flag, opStr);
49,908✔
3086
    int32_t code = dumpBlockData(pBlock, flagBuf, &pBuf, taskIdStr);
49,908✔
3087
    if (code == 0) {
49,907!
3088
      qDebug("%s", pBuf);
49,907!
3089
      taosMemoryFree(pBuf);
49,905!
3090
    }
3091
  }
3092
}
3093

3094
TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }
4,836,529!
3095

3096
void updateTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pWin, int64_t delta) {
662,730,063✔
3097
  int64_t* ts = (int64_t*)pColData->pData;
662,730,063✔
3098

3099
  int64_t duration = pWin->ekey - pWin->skey + delta;
662,730,063✔
3100
  ts[2] = duration;            // set the duration
662,730,063✔
3101
  ts[3] = pWin->skey;          // window start key
662,730,063✔
3102
  ts[4] = pWin->ekey + delta;  // window end key
662,730,063✔
3103
}
662,730,063✔
3104

3105
int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t oldKeysLen, const SSDataBlock* pBlock,
21,377,778✔
3106
                 int32_t rowIndex) {
3107
  SColumnDataAgg* pColAgg = NULL;
21,377,778✔
3108
  const char*     isNull = oldkeyBuf;
21,377,778✔
3109
  const char*     p = oldkeyBuf + sizeof(int8_t) * pSortGroupCols->size;
21,377,778✔
3110

3111
  for (int32_t i = 0; i < pSortGroupCols->size; ++i) {
53,642,593✔
3112
    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
32,747,466✔
3113
    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
32,747,466✔
3114
    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
32,747,466!
3115

3116
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
65,494,932!
3117
      if (isNull[i] != 1) return 1;
2,183,033✔
3118
    } else {
3119
      if (isNull[i] != 0) return 1;
30,564,433✔
3120
      const char* val = colDataGetData(pColInfoData, rowIndex);
30,550,210!
3121
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
30,550,210!
3122
        int32_t len = getJsonValueLen(val);
×
3123
        if (memcmp(p, val, len) != 0) return 1;
×
3124
        p += len;
×
3125
      } else if (IS_VAR_DATA_TYPE(pCol->type)) {
30,550,210!
3126
        if (memcmp(p, val, varDataTLen(val)) != 0) return 1;
10,385,405✔
3127
        p += varDataTLen(val);
10,249,144✔
3128
      } else {
3129
        if (0 != memcmp(p, val, pCol->bytes)) return 1;
20,164,805✔
3130
        p += pCol->bytes;
19,832,830✔
3131
      }
3132
    }
3133
  }
3134
  if ((int32_t)(p - oldkeyBuf) != oldKeysLen) return 1;
20,895,127✔
3135
  return 0;
20,894,831✔
3136
}
3137

3138
int32_t buildKeys(char* keyBuf, const SArray* pSortGroupCols, const SSDataBlock* pBlock, int32_t rowIndex) {
482,519✔
3139
  uint32_t        colNum = pSortGroupCols->size;
482,519✔
3140
  SColumnDataAgg* pColAgg = NULL;
482,519✔
3141
  char*           isNull = keyBuf;
482,519✔
3142
  char*           p = keyBuf + sizeof(int8_t) * colNum;
482,519✔
3143

3144
  for (int32_t i = 0; i < colNum; ++i) {
1,436,928✔
3145
    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
954,409✔
3146
    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
954,409✔
3147
    if (pCol->slotId > pBlock->pDataBlock->size) continue;
954,409!
3148

3149
    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
954,409!
3150

3151
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
1,908,818!
3152
      isNull[i] = 1;
34,548✔
3153
    } else {
3154
      isNull[i] = 0;
919,861✔
3155
      const char* val = colDataGetData(pColInfoData, rowIndex);
919,861!
3156
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
919,861!
3157
        int32_t len = getJsonValueLen(val);
×
3158
        memcpy(p, val, len);
×
3159
        p += len;
×
3160
      } else if (IS_VAR_DATA_TYPE(pCol->type)) {
919,861!
3161
        varDataCopy(p, val);
139,601✔
3162
        p += varDataTLen(val);
139,601✔
3163
      } else {
3164
        memcpy(p, val, pCol->bytes);
780,260✔
3165
        p += pCol->bytes;
780,260✔
3166
      }
3167
    }
3168
  }
3169
  return (int32_t)(p - keyBuf);
482,519✔
3170
}
3171

3172
uint64_t calcGroupId(char* pData, int32_t len) {
40,758,646✔
3173
  T_MD5_CTX context;
3174
  tMD5Init(&context);
40,758,646✔
3175
  tMD5Update(&context, (uint8_t*)pData, len);
40,726,447✔
3176
  tMD5Final(&context);
40,869,749✔
3177

3178
  // NOTE: only extract the initial 8 bytes of the final MD5 digest
3179
  uint64_t id = 0;
41,014,623✔
3180
  memcpy(&id, context.digest, sizeof(uint64_t));
41,014,623✔
3181
  if (0 == id) memcpy(&id, context.digest + 8, sizeof(uint64_t));
41,014,623!
3182
  return id;
41,014,623✔
3183
}
3184

3185
SNodeList* makeColsNodeArrFromSortKeys(SNodeList* pSortKeys) {
944✔
3186
  SNode*     node;
3187
  SNodeList* ret = NULL;
944✔
3188
  FOREACH(node, pSortKeys) {
2,879!
3189
    SOrderByExprNode* pSortKey = (SOrderByExprNode*)node;
1,935✔
3190
    int32_t           code = nodesListMakeAppend(&ret, pSortKey->pExpr);
1,935✔
3191
    if (code != TSDB_CODE_SUCCESS) {
1,935!
3192
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
3193
      terrno = code;
×
3194
      return NULL;
×
3195
    }
3196
  }
3197
  return ret;
944✔
3198
}
3199

3200
int32_t extractKeysLen(const SArray* keys, int32_t* pLen) {
944✔
3201
  int32_t code = TSDB_CODE_SUCCESS;
944✔
3202
  int32_t lino = 0;
944✔
3203
  int32_t len = 0;
944✔
3204
  int32_t keyNum = taosArrayGetSize(keys);
944✔
3205
  for (int32_t i = 0; i < keyNum; ++i) {
2,396✔
3206
    SColumn* pCol = (SColumn*)taosArrayGet(keys, i);
1,452✔
3207
    QUERY_CHECK_NULL(pCol, code, lino, _end, terrno);
1,452!
3208
    len += pCol->bytes;
1,452✔
3209
  }
3210
  len += sizeof(int8_t) * keyNum;  // null flag
944✔
3211
  *pLen = len;
944✔
3212

3213
_end:
944✔
3214
  if (code != TSDB_CODE_SUCCESS) {
944!
3215
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
3216
  }
3217
  return code;
942✔
3218
}
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