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

taosdata / TDengine / #3678

22 Mar 2025 04:46PM UTC coverage: 62.181% (+0.4%) from 61.745%
#3678

push

travis-ci

web-flow
fix: userOperTest in linux (#30363)

Co-authored-by: taos-support <it@taosdata.com>

154967 of 318145 branches covered (48.71%)

Branch coverage included in aggregate %.

240358 of 317625 relevant lines covered (75.67%)

16255149.31 hits per line

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

67.77
/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, uint64_t suid);
55

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

59
static int64_t getLimit(const SNode* pLimit) { return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->limit) ? -1 : ((SLimitNode*)pLimit)->limit->datum.i; }
28,231,957!
60
static int64_t getOffset(const SNode* pLimit) { return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->offset) ? -1 : ((SLimitNode*)pLimit)->offset->datum.i; }
28,238,292✔
61
static void    releaseColInfoData(void* pCol);
62

63
void initResultRowInfo(SResultRowInfo* pResultRowInfo) {
11,726,347✔
64
  pResultRowInfo->size = 0;
11,726,347✔
65
  pResultRowInfo->cur.pageId = -1;
11,726,347✔
66
}
11,726,347✔
67

68
void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }
8,538,438✔
69

70
void resetResultRow(SResultRow* pResultRow, size_t entrySize) {
71,660,260✔
71
  pResultRow->numOfRows = 0;
71,660,260✔
72
  pResultRow->closed = false;
71,660,260✔
73
  pResultRow->endInterp = false;
71,660,260✔
74
  pResultRow->startInterp = false;
71,660,260✔
75

76
  if (entrySize > 0) {
71,660,260!
77
    memset(pResultRow->pEntryInfo, 0, entrySize);
71,671,028✔
78
  }
79
}
71,660,260✔
80

81
// TODO refactor: use macro
82
SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) {
2,147,483,647✔
83
  return (SResultRowEntryInfo*)((char*)pRow->pEntryInfo + offset[index]);
2,147,483,647✔
84
}
85

86
size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
8,251,823✔
87
  int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow);
8,251,823✔
88

89
  for (int32_t i = 0; i < numOfOutput; ++i) {
33,224,064✔
90
    rowSize += pCtx[i].resDataInfo.interBufSize;
24,972,241✔
91
  }
92

93
  return rowSize;
8,251,823✔
94
}
95

96
// Convert buf read from rocksdb to result row
97
int32_t getResultRowFromBuf(SExprSupp *pSup, const char* inBuf, size_t inBufSize, char **outBuf, size_t *outBufSize) {
2,738✔
98
  if (inBuf == NULL || pSup == NULL) {
2,738!
99
    qError("invalid input parameters, inBuf:%p, pSup:%p", inBuf, pSup);
10!
100
    return TSDB_CODE_INVALID_PARA;
10✔
101
  }
102
  SqlFunctionCtx *pCtx = pSup->pCtx;
2,728✔
103
  int32_t        *offset = pSup->rowEntryInfoOffset;
2,728✔
104
  SResultRow     *pResultRow  = NULL;
2,728✔
105
  size_t          processedSize = 0;
2,728✔
106
  int32_t         code = TSDB_CODE_SUCCESS;
2,728✔
107

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

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

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

153
// Convert result row to buf for rocksdb
154
int32_t putResultRowToBuf(SExprSupp *pSup, const char* inBuf, size_t inBufSize, char **outBuf, size_t *outBufSize) {
1,428,741✔
155
  if (pSup == NULL || inBuf == NULL || outBuf == NULL || outBufSize == NULL) {
1,428,741!
156
    qError("invalid input parameters, inBuf:%p, pSup:%p, outBufSize:%p, outBuf:%p", inBuf, pSup, outBufSize, outBuf);
×
157
    return TSDB_CODE_INVALID_PARA;
×
158
  }
159

160
  SqlFunctionCtx *pCtx = pSup->pCtx;
1,428,741✔
161
  int32_t        *offset = pSup->rowEntryInfoOffset;
1,428,741✔
162
  SResultRow     *pResultRow = (SResultRow*)inBuf;
1,428,741✔
163
  size_t          rowSize = getResultRowSize(pCtx, pSup->numOfExprs);
1,428,741✔
164

165
  if (rowSize > inBufSize) {
1,428,741!
166
    qError("invalid input buffer size, rowSize:%zu, inBufSize:%zu", rowSize, inBufSize);
×
167
    return TSDB_CODE_INVALID_PARA;
×
168
  }
169

170
  // calculate the size of output buffer
171
  *outBufSize = rowSize + sizeof(int32_t) * pSup->numOfExprs;
1,428,741✔
172
  if (rowSize < inBufSize) {
1,428,741✔
173
    *outBufSize += inBufSize - rowSize;
749✔
174
  }
175

176
  *outBuf = taosMemoryMalloc(*outBufSize);
1,428,741!
177
  if (*outBuf == NULL) {
1,428,741!
178
    qError("failed to allocate memory for output buffer, size:%zu", *outBufSize);
×
179
    return terrno;
×
180
  }
181

182
  char *pBuf = *outBuf;
1,428,741✔
183
  pResultRow->version = FUNCTION_RESULT_INFO_VERSION;
1,428,741✔
184
  (void)memcpy(pBuf, pResultRow, sizeof(SResultRow));
1,428,741✔
185
  pBuf += sizeof(SResultRow);
1,428,741✔
186
  for (int32_t i = 0; i < pSup->numOfExprs; ++i) {
5,778,741✔
187
    size_t len = sizeof(SResultRowEntryInfo) + pCtx[i].resDataInfo.interBufSize;
4,350,000✔
188
    *(int32_t *) pBuf = (int32_t)len;
4,350,000✔
189
    pBuf += sizeof(int32_t);
4,350,000✔
190
    (void)memcpy(pBuf, getResultEntryInfo(pResultRow, i, offset), len);
4,350,000✔
191
    pBuf += len;
4,350,000✔
192
  }
193

194
  if (rowSize < inBufSize) {
1,428,741✔
195
    // stream stores extra data after result row
196
    size_t leftLen = inBufSize - rowSize;
749✔
197
    (void)memcpy(pBuf, inBuf + rowSize, leftLen);
749✔
198
    pBuf += leftLen;
749✔
199
  }
200
  return TSDB_CODE_SUCCESS;
1,428,741✔
201
}
202

203
static void freeEx(void* p) { taosMemoryFree(*(void**)p); }
×
204

205
void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
4,534,457✔
206
  taosMemoryFreeClear(pGroupResInfo->pBuf);
4,534,457!
207
  if (pGroupResInfo->freeItem) {
4,535,501!
208
    //    taosArrayDestroy(pGroupResInfo->pRows);
209
    taosArrayDestroyEx(pGroupResInfo->pRows, freeEx);
×
210
    pGroupResInfo->freeItem = false;
×
211
    pGroupResInfo->pRows = NULL;
×
212
  } else {
213
    taosArrayDestroy(pGroupResInfo->pRows);
4,535,501✔
214
    pGroupResInfo->pRows = NULL;
4,536,556✔
215
  }
216
  pGroupResInfo->index = 0;
4,536,556✔
217
  pGroupResInfo->delIndex = 0;
4,536,556✔
218
}
4,536,556✔
219

220
int32_t resultrowComparAsc(const void* p1, const void* p2) {
1,968,967,144✔
221
  SResKeyPos* pp1 = *(SResKeyPos**)p1;
1,968,967,144✔
222
  SResKeyPos* pp2 = *(SResKeyPos**)p2;
1,968,967,144✔
223

224
  if (pp1->groupId == pp2->groupId) {
1,968,967,144✔
225
    int64_t pts1 = *(int64_t*)pp1->key;
1,788,766,683✔
226
    int64_t pts2 = *(int64_t*)pp2->key;
1,788,766,683✔
227

228
    if (pts1 == pts2) {
1,788,766,683!
229
      return 0;
×
230
    } else {
231
      return pts1 < pts2 ? -1 : 1;
1,788,766,683✔
232
    }
233
  } else {
234
    return pp1->groupId < pp2->groupId ? -1 : 1;
180,200,461✔
235
  }
236
}
237

238
static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); }
74,366,932✔
239

240
int32_t initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order) {
3,626,190✔
241
  int32_t code = TSDB_CODE_SUCCESS;
3,626,190✔
242
  int32_t lino = 0;
3,626,190✔
243
  if (pGroupResInfo->pRows != NULL) {
3,626,190✔
244
    taosArrayDestroy(pGroupResInfo->pRows);
49,918✔
245
  }
246
  if (pGroupResInfo->pBuf) {
3,626,190✔
247
    taosMemoryFree(pGroupResInfo->pBuf);
49,918!
248
    pGroupResInfo->pBuf = NULL;
49,918✔
249
  }
250

251
  // extract the result rows information from the hash map
252
  int32_t size = tSimpleHashGetSize(pHashmap);
3,626,190✔
253

254
  void* pData = NULL;
3,626,645✔
255
  pGroupResInfo->pRows = taosArrayInit(size, POINTER_BYTES);
3,626,645✔
256
  QUERY_CHECK_NULL(pGroupResInfo->pRows, code, lino, _end, terrno);
3,627,736!
257

258
  size_t  keyLen = 0;
3,627,736✔
259
  int32_t iter = 0;
3,627,736✔
260
  int64_t bufLen = 0, offset = 0;
3,627,736✔
261

262
  // todo move away and record this during create window
263
  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
218,007,100✔
264
    /*void* key = */ (void)tSimpleHashGetKey(pData, &keyLen);
265
    bufLen += keyLen + sizeof(SResultRowPosition);
214,379,364✔
266
  }
267

268
  pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);
3,615,342!
269
  QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
3,627,775!
270

271
  iter = 0;
3,627,775✔
272
  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
217,852,142✔
273
    void* key = tSimpleHashGetKey(pData, &keyLen);
214,239,404✔
274

275
    SResKeyPos* p = (SResKeyPos*)(pGroupResInfo->pBuf + offset);
214,239,404✔
276

277
    p->groupId = *(uint64_t*)key;
214,239,404✔
278
    p->pos = *(SResultRowPosition*)pData;
214,239,404✔
279
    memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t));
214,239,404✔
280
    void* tmp = taosArrayPush(pGroupResInfo->pRows, &p);
214,239,404✔
281
    QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
214,224,367!
282

283
    offset += keyLen + sizeof(struct SResultRowPosition);
214,224,367✔
284
  }
285

286
  if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
3,596,840✔
287
    __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc;
1,655,319✔
288
    size = POINTER_BYTES;
1,655,319✔
289
    taosSort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), size, fn);
1,655,319✔
290
  }
291

292
  pGroupResInfo->index = 0;
3,628,067✔
293

294
_end:
3,628,067✔
295
  if (code != TSDB_CODE_SUCCESS) {
3,628,067!
296
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
297
  }
298
  return code;
3,627,126✔
299
}
300

301
void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList) {
34,855✔
302
  if (pGroupResInfo->pRows != NULL) {
34,855✔
303
    taosArrayDestroy(pGroupResInfo->pRows);
31,077✔
304
  }
305

306
  pGroupResInfo->freeItem = true;
34,856✔
307
  pGroupResInfo->pRows = pArrayList;
34,856✔
308
  pGroupResInfo->index = 0;
34,856✔
309
  pGroupResInfo->delIndex = 0;
34,856✔
310
}
34,856✔
311

312
bool hasRemainResults(SGroupResInfo* pGroupResInfo) {
10,404,243✔
313
  if (pGroupResInfo->pRows == NULL) {
10,404,243✔
314
    return false;
1,401✔
315
  }
316

317
  return pGroupResInfo->index < taosArrayGetSize(pGroupResInfo->pRows);
10,402,842✔
318
}
319

320
int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {
6,904,280✔
321
  if (pGroupResInfo->pRows == 0) {
6,904,280✔
322
    return 0;
2,155✔
323
  }
324

325
  return (int32_t)taosArrayGetSize(pGroupResInfo->pRows);
6,902,125✔
326
}
327

328
SArray* createSortInfo(SNodeList* pNodeList) {
2,321,317✔
329
  size_t numOfCols = 0;
2,321,317✔
330

331
  if (pNodeList != NULL) {
2,321,317✔
332
    numOfCols = LIST_LENGTH(pNodeList);
2,319,601!
333
  } else {
334
    numOfCols = 0;
1,716✔
335
  }
336

337
  SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));
2,321,317✔
338
  if (pList == NULL) {
2,321,344!
339
    return pList;
×
340
  }
341

342
  for (int32_t i = 0; i < numOfCols; ++i) {
4,794,840✔
343
    SOrderByExprNode* pSortKey = (SOrderByExprNode*)nodesListGetNode(pNodeList, i);
2,473,409✔
344
    if (!pSortKey) {
2,473,473✔
345
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
18!
346
      taosArrayDestroy(pList);
18✔
347
      pList = NULL;
×
348
      terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
349
      break;
×
350
    }
351
    SBlockOrderInfo   bi = {0};
2,473,455✔
352
    bi.order = (pSortKey->order == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
2,473,455✔
353
    bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST);
2,473,455✔
354

355
    SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr;
2,473,455✔
356
    bi.slotId = pColNode->slotId;
2,473,455✔
357
    void* tmp = taosArrayPush(pList, &bi);
2,473,496✔
358
    if (!tmp) {
2,473,496!
359
      taosArrayDestroy(pList);
×
360
      pList = NULL;
×
361
      break;
×
362
    }
363
  }
364

365
  return pList;
2,321,431✔
366
}
367

368
SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {
21,204,678✔
369
  int32_t      numOfCols = LIST_LENGTH(pNode->pSlots);
21,204,678!
370
  SSDataBlock* pBlock = NULL;
21,204,678✔
371
  int32_t      code = createDataBlock(&pBlock);
21,204,678✔
372
  if (code) {
21,218,689!
373
    terrno = code;
×
374
    return NULL;
×
375
  }
376

377
  pBlock->info.id.blockId = pNode->dataBlockId;
21,222,410✔
378
  pBlock->info.type = STREAM_INVALID;
21,222,410✔
379
  pBlock->info.calWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
21,222,410✔
380
  pBlock->info.watermark = INT64_MIN;
21,222,410✔
381

382
  for (int32_t i = 0; i < numOfCols; ++i) {
115,888,175✔
383
    SSlotDescNode*  pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
94,650,406✔
384
    if (!pDescNode) {
94,732,930!
385
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
386
      blockDataDestroy(pBlock);
×
387
      pBlock = NULL;
×
388
      terrno = TSDB_CODE_INVALID_PARA;
×
389
      break;
×
390
    }
391
    SColumnInfoData idata =
392
        createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);
94,732,930✔
393
    idata.info.scale = pDescNode->dataType.scale;
94,830,620✔
394
    idata.info.precision = pDescNode->dataType.precision;
94,830,620✔
395
    idata.info.noData = pDescNode->reserve;
94,830,620✔
396

397
    code = blockDataAppendColInfo(pBlock, &idata);
94,830,620✔
398
    if (code != TSDB_CODE_SUCCESS) {
94,665,765!
399
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
400
      blockDataDestroy(pBlock);
×
401
      pBlock = NULL;
×
402
      terrno = code;
×
403
      break;
×
404
    }
405
  }
406

407
  return pBlock;
21,237,769✔
408
}
409

410
int32_t prepareDataBlockBuf(SSDataBlock* pDataBlock, SColMatchInfo* pMatchInfo) {
4,898,361✔
411
  SDataBlockInfo* pBlockInfo = &pDataBlock->info;
4,898,361✔
412

413
  for (int32_t i = 0; i < taosArrayGetSize(pMatchInfo->pList); ++i) {
28,917,489✔
414
    SColMatchItem* pItem = taosArrayGet(pMatchInfo->pList, i);
24,122,231✔
415
    if (!pItem) {
24,134,606✔
416
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
3,087!
417
      return terrno;
3,087✔
418
    }
419

420
    if (pItem->isPk) {
24,131,519✔
421
      SColumnInfoData* pInfoData = taosArrayGet(pDataBlock->pDataBlock, pItem->dstSlotId);
112,391✔
422
      if (!pInfoData) {
112,405!
423
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
424
        return terrno;
×
425
      }
426
      pBlockInfo->pks[0].type = pInfoData->info.type;
112,405✔
427
      pBlockInfo->pks[1].type = pInfoData->info.type;
112,405✔
428

429
      // allocate enough buffer size, which is pInfoData->info.bytes
430
      if (IS_VAR_DATA_TYPE(pItem->dataType.type)) {
112,405!
431
        pBlockInfo->pks[0].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
31,321!
432
        if (pBlockInfo->pks[0].pData == NULL) {
31,216!
433
          return terrno;
×
434
        }
435

436
        pBlockInfo->pks[1].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
31,216!
437
        if (pBlockInfo->pks[1].pData == NULL) {
31,226!
438
          taosMemoryFreeClear(pBlockInfo->pks[0].pData);
×
439
          return terrno;
×
440
        }
441

442
        pBlockInfo->pks[0].nData = pInfoData->info.bytes;
31,226✔
443
        pBlockInfo->pks[1].nData = pInfoData->info.bytes;
31,226✔
444
      }
445

446
      break;
112,310✔
447
    }
448
  }
449

450
  return TSDB_CODE_SUCCESS;
4,897,036✔
451
}
452

453
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
6,783✔
454
  STransTagExprCtx* pCtx = pContext;
6,783✔
455
  SMetaReader*      mr = pCtx->pReader;
6,783✔
456
  bool              isTagCol = false, isTbname = false;
6,783✔
457
  if (nodeType(*pNode) == QUERY_NODE_COLUMN) {
6,783✔
458
    SColumnNode* pCol = (SColumnNode*)*pNode;
1,951✔
459
    if (pCol->colType == COLUMN_TYPE_TBNAME)
1,951!
460
      isTbname = true;
×
461
    else
462
      isTagCol = true;
1,951✔
463
  } else if (nodeType(*pNode) == QUERY_NODE_FUNCTION) {
4,832!
464
    SFunctionNode* pFunc = (SFunctionNode*)*pNode;
×
465
    if (pFunc->funcType == FUNCTION_TYPE_TBNAME) isTbname = true;
×
466
  }
467
  if (isTagCol) {
6,783✔
468
    SColumnNode* pSColumnNode = *(SColumnNode**)pNode;
1,951✔
469

470
    SValueNode* res = NULL;
1,951✔
471
    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
1,951✔
472
    if (NULL == res) {
1,951!
473
      return DEAL_RES_ERROR;
×
474
    }
475

476
    res->translate = true;
1,951✔
477
    res->node.resType = pSColumnNode->node.resType;
1,951✔
478

479
    STagVal tagVal = {0};
1,951✔
480
    tagVal.cid = pSColumnNode->colId;
1,951✔
481
    const char* p = mr->pAPI->extractTagVal(mr->me.ctbEntry.pTags, pSColumnNode->node.resType.type, &tagVal);
1,951✔
482
    if (p == NULL) {
1,951!
483
      res->node.resType.type = TSDB_DATA_TYPE_NULL;
×
484
    } else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {
1,951!
485
      int32_t len = ((const STag*)p)->len;
×
486
      res->datum.p = taosMemoryCalloc(len + 1, 1);
×
487
      if (NULL == res->datum.p) {
×
488
        return DEAL_RES_ERROR;
×
489
      }
490
      memcpy(res->datum.p, p, len);
×
491
    } else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) {
1,951!
492
      res->datum.p = taosMemoryCalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1, 1);
1,928!
493
      if (NULL == res->datum.p) {
1,928!
494
        return DEAL_RES_ERROR;
×
495
      }
496
      memcpy(varDataVal(res->datum.p), tagVal.pData, tagVal.nData);
1,928✔
497
      varDataSetLen(res->datum.p, tagVal.nData);
1,928✔
498
    } else {
499
      int32_t code = nodesSetValueNodeValue(res, &(tagVal.i64));
23✔
500
      if (code != TSDB_CODE_SUCCESS) {
23!
501
        return DEAL_RES_ERROR;
×
502
      }
503
    }
504
    nodesDestroyNode(*pNode);
1,951✔
505
    *pNode = (SNode*)res;
1,951✔
506
  } else if (isTbname) {
4,832!
507
    SValueNode* res = NULL;
×
508
    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
×
509
    if (NULL == res) {
×
510
      return DEAL_RES_ERROR;
×
511
    }
512

513
    res->translate = true;
×
514
    res->node.resType = ((SExprNode*)(*pNode))->resType;
×
515

516
    int32_t len = strlen(mr->me.name);
×
517
    res->datum.p = taosMemoryCalloc(len + VARSTR_HEADER_SIZE + 1, 1);
×
518
    if (NULL == res->datum.p) {
×
519
      return DEAL_RES_ERROR;
×
520
    }
521
    memcpy(varDataVal(res->datum.p), mr->me.name, len);
×
522
    varDataSetLen(res->datum.p, len);
×
523
    nodesDestroyNode(*pNode);
×
524
    *pNode = (SNode*)res;
×
525
  }
526

527
  return DEAL_RES_CONTINUE;
6,783✔
528
}
529

530
int32_t isQualifiedTable(STableKeyInfo* info, SNode* pTagCond, void* metaHandle, bool* pQualified, SStorageAPI* pAPI) {
973✔
531
  int32_t     code = TSDB_CODE_SUCCESS;
973✔
532
  SMetaReader mr = {0};
973✔
533

534
  pAPI->metaReaderFn.initReader(&mr, metaHandle, META_READER_LOCK, &pAPI->metaFn);
973✔
535
  code = pAPI->metaReaderFn.getEntryGetUidCache(&mr, info->uid);
973✔
536
  if (TSDB_CODE_SUCCESS != code) {
973!
537
    pAPI->metaReaderFn.clearReader(&mr);
×
538
    *pQualified = false;
×
539

540
    return TSDB_CODE_SUCCESS;
×
541
  }
542

543
  SNode* pTagCondTmp = NULL;
973✔
544
  code = nodesCloneNode(pTagCond, &pTagCondTmp);
973✔
545
  if (TSDB_CODE_SUCCESS != code) {
973!
546
    *pQualified = false;
×
547
    pAPI->metaReaderFn.clearReader(&mr);
×
548
    return code;
×
549
  }
550
  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
973✔
551
  nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &ctx);
973✔
552
  pAPI->metaReaderFn.clearReader(&mr);
973✔
553
  if (TSDB_CODE_SUCCESS != ctx.code) {
973!
554
    *pQualified = false;
×
555
    terrno = code;
×
556
    return code;
×
557
  }
558

559
  SNode* pNew = NULL;
973✔
560
  code = scalarCalculateConstants(pTagCondTmp, &pNew);
973✔
561
  if (TSDB_CODE_SUCCESS != code) {
973!
562
    terrno = code;
×
563
    nodesDestroyNode(pTagCondTmp);
×
564
    *pQualified = false;
×
565

566
    return code;
×
567
  }
568

569
  SValueNode* pValue = (SValueNode*)pNew;
973✔
570
  *pQualified = pValue->datum.b;
973✔
571

572
  nodesDestroyNode(pNew);
973✔
573
  return TSDB_CODE_SUCCESS;
973✔
574
}
575

576
static EDealRes getColumn(SNode** pNode, void* pContext) {
50,242,938✔
577
  tagFilterAssist* pData = (tagFilterAssist*)pContext;
50,242,938✔
578
  SColumnNode*     pSColumnNode = NULL;
50,242,938✔
579
  if (QUERY_NODE_COLUMN == nodeType((*pNode))) {
50,242,938✔
580
    pSColumnNode = *(SColumnNode**)pNode;
3,470,617✔
581
  } else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {
46,772,321✔
582
    SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);
479,353✔
583
    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {
479,353✔
584
      pData->code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pSColumnNode);
468,310✔
585
      if (NULL == pSColumnNode) {
468,304!
586
        return DEAL_RES_ERROR;
×
587
      }
588
      pSColumnNode->colId = -1;
468,304✔
589
      pSColumnNode->colType = COLUMN_TYPE_TBNAME;
468,304✔
590
      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;
468,304✔
591
      pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;
468,304✔
592
      nodesDestroyNode(*pNode);
468,304✔
593
      *pNode = (SNode*)pSColumnNode;
468,311✔
594
    } else {
595
      return DEAL_RES_CONTINUE;
11,043✔
596
    }
597
  } else {
598
    return DEAL_RES_CONTINUE;
46,292,968✔
599
  }
600

601
  void* data = taosHashGet(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));
3,938,928✔
602
  if (!data) {
3,936,844✔
603
    int32_t tempRes =
604
        taosHashPut(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));
3,271,544✔
605
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
3,276,799!
606
      return DEAL_RES_ERROR;
×
607
    }
608
    pSColumnNode->slotId = pData->index++;
3,276,799✔
609
    SColumnInfo cInfo = {.colId = pSColumnNode->colId,
3,276,799✔
610
                         .type = pSColumnNode->node.resType.type,
3,276,799✔
611
                         .bytes = pSColumnNode->node.resType.bytes,
3,276,799✔
612
                         .pk = pSColumnNode->isPk};
3,276,799✔
613
#if TAG_FILTER_DEBUG
614
    qDebug("tagfilter build column info, slotId:%d, colId:%d, type:%d", pSColumnNode->slotId, cInfo.colId, cInfo.type);
615
#endif
616
    void* tmp = taosArrayPush(pData->cInfoList, &cInfo);
3,276,799✔
617
    if (!tmp) {
3,274,871!
618
      return DEAL_RES_ERROR;
×
619
    }
620
  } else {
621
    SColumnNode* col = *(SColumnNode**)data;
665,300✔
622
    pSColumnNode->slotId = col->slotId;
665,300✔
623
  }
624

625
  return DEAL_RES_CONTINUE;
3,940,171✔
626
}
627

628
static int32_t createResultData(SDataType* pType, int32_t numOfRows, SScalarParam* pParam) {
1,447,941✔
629
  SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
1,447,941!
630
  if (pColumnData == NULL) {
1,448,808!
631
    return terrno;
×
632
  }
633

634
  pColumnData->info.type = pType->type;
1,448,808✔
635
  pColumnData->info.bytes = pType->bytes;
1,448,808✔
636
  pColumnData->info.scale = pType->scale;
1,448,808✔
637
  pColumnData->info.precision = pType->precision;
1,448,808✔
638

639
  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);
1,448,808✔
640
  if (code != TSDB_CODE_SUCCESS) {
1,448,514!
641
    terrno = code;
×
642
    releaseColInfoData(pColumnData);
×
643
    return terrno;
×
644
  }
645

646
  pParam->columnData = pColumnData;
1,448,547✔
647
  pParam->colAlloced = true;
1,448,547✔
648
  return TSDB_CODE_SUCCESS;
1,448,547✔
649
}
650

651
static void releaseColInfoData(void* pCol) {
77,158✔
652
  if (pCol) {
77,158!
653
    SColumnInfoData* col = (SColumnInfoData*)pCol;
77,164✔
654
    colDataDestroy(col);
77,164✔
655
    taosMemoryFree(col);
77,176!
656
  }
657
}
77,173✔
658

659
void freeItem(void* p) {
3,452,786✔
660
  STUidTagInfo* pInfo = p;
3,452,786✔
661
  if (pInfo->pTagVal != NULL) {
3,452,786✔
662
    taosMemoryFree(pInfo->pTagVal);
3,264,901!
663
  }
664
}
3,453,894✔
665

666
static int32_t genTagFilterDigest(const SNode* pTagCond, T_MD5_CTX* pContext) {
×
667
  if (pTagCond == NULL) {
×
668
    return TSDB_CODE_SUCCESS;
×
669
  }
670

671
  char*   payload = NULL;
×
672
  int32_t len = 0;
×
673
  int32_t code = nodesNodeToMsg(pTagCond, &payload, &len);
×
674
  if (code != TSDB_CODE_SUCCESS) {
×
675
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
676
    return code;
×
677
  }
678

679
  tMD5Init(pContext);
×
680
  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
×
681
  tMD5Final(pContext);
×
682

683
  taosMemoryFree(payload);
×
684
  return TSDB_CODE_SUCCESS;
×
685
}
686

687
static int32_t genTbGroupDigest(const SNode* pGroup, uint8_t* filterDigest, T_MD5_CTX* pContext) {
×
688
  int32_t code = TSDB_CODE_SUCCESS;
×
689
  int32_t lino = 0;
×
690
  char*   payload = NULL;
×
691
  int32_t len = 0;
×
692
  code = nodesNodeToMsg(pGroup, &payload, &len);
×
693
  QUERY_CHECK_CODE(code, lino, _end);
×
694

695
  if (filterDigest[0]) {
×
696
    payload = taosMemoryRealloc(payload, len + tListLen(pContext->digest));
×
697
    QUERY_CHECK_NULL(payload, code, lino, _end, terrno);
×
698
    memcpy(payload + len, filterDigest + 1, tListLen(pContext->digest));
×
699
    len += tListLen(pContext->digest);
×
700
  }
701

702
  tMD5Init(pContext);
×
703
  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
×
704
  tMD5Final(pContext);
×
705

706
_end:
×
707
  taosMemoryFree(payload);
×
708
  if (code != TSDB_CODE_SUCCESS) {
×
709
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
710
  }
711
  return code;
×
712
}
713

714
int32_t getColInfoResultForGroupby(void* pVnode, SNodeList* group, STableListInfo* pTableListInfo, uint8_t* digest,
65,617✔
715
                                   SStorageAPI* pAPI, bool initRemainGroups) {
716
  int32_t      code = TSDB_CODE_SUCCESS;
65,617✔
717
  int32_t      lino = 0;
65,617✔
718
  SArray*      pBlockList = NULL;
65,617✔
719
  SSDataBlock* pResBlock = NULL;
65,617✔
720
  void*        keyBuf = NULL;
65,617✔
721
  SArray*      groupData = NULL;
65,617✔
722
  SArray*      pUidTagList = NULL;
65,617✔
723
  SArray*      tableList = NULL;
65,617✔
724

725
  int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);
65,617✔
726
  if (rows == 0) {
65,620!
727
    return TSDB_CODE_SUCCESS;
×
728
  }
729

730
  tagFilterAssist ctx = {0};
65,620✔
731
  ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
65,620✔
732
  if (ctx.colHash == NULL) {
65,697!
733
    code = terrno;
×
734
    goto end;
×
735
  }
736

737
  ctx.index = 0;
65,697✔
738
  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
65,697✔
739
  if (ctx.cInfoList == NULL) {
65,732✔
740
    code = terrno;
1✔
741
    goto end;
×
742
  }
743

744
  SNode* pNode = NULL;
65,731✔
745
  FOREACH(pNode, group) {
142,908!
746
    nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);
77,162✔
747
    if (TSDB_CODE_SUCCESS != ctx.code) {
77,177!
748
      code = ctx.code;
×
749
      goto end;
×
750
    }
751
    REPLACE_NODE(pNode);
77,177✔
752
  }
753

754
  T_MD5_CTX context = {0};
65,746✔
755
  if (tsTagFilterCache) {
65,746!
756
    SNodeListNode* listNode = NULL;
×
757
    code = nodesMakeNode(QUERY_NODE_NODE_LIST, (SNode**)&listNode);
×
758
    if (TSDB_CODE_SUCCESS != code) {
×
759
      goto end;
×
760
    }
761
    listNode->pNodeList = group;
×
762
    code = genTbGroupDigest((SNode*)listNode, digest, &context);
×
763
    QUERY_CHECK_CODE(code, lino, end);
×
764

765
    nodesFree(listNode);
×
766

767
    code = pAPI->metaFn.metaGetCachedTbGroup(pVnode, pTableListInfo->idInfo.suid, context.digest,
×
768
                                             tListLen(context.digest), &tableList);
769
    QUERY_CHECK_CODE(code, lino, end);
×
770

771
    if (tableList) {
×
772
      taosArrayDestroy(pTableListInfo->pTableList);
×
773
      pTableListInfo->pTableList = tableList;
×
774
      qDebug("retrieve tb group list from cache, numOfTables:%d",
×
775
             (int32_t)taosArrayGetSize(pTableListInfo->pTableList));
776
      goto end;
×
777
    }
778
  }
779

780
  pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));
65,746✔
781
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
65,658!
782

783
  for (int32_t i = 0; i < rows; ++i) {
325,510✔
784
    STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
259,744✔
785
    QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);
259,677!
786
    STUidTagInfo   info = {.uid = pkeyInfo->uid};
259,677✔
787
    void*          tmp = taosArrayPush(pUidTagList, &info);
259,852✔
788
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
259,852!
789
  }
790

791
  code = pAPI->metaFn.getTableTags(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
65,766✔
792
  if (code != TSDB_CODE_SUCCESS) {
65,726!
793
    goto end;
×
794
  }
795

796
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
65,726✔
797
  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);
65,711✔
798
  if (pResBlock == NULL) {
65,724!
799
    code = terrno;
×
800
    goto end;
×
801
  }
802

803
  //  int64_t st1 = taosGetTimestampUs();
804
  //  qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
805

806
  pBlockList = taosArrayInit(2, POINTER_BYTES);
65,724✔
807
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
65,679!
808

809
  void* tmp = taosArrayPush(pBlockList, &pResBlock);
65,698✔
810
  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
65,698!
811

812
  groupData = taosArrayInit(2, POINTER_BYTES);
65,698✔
813
  QUERY_CHECK_NULL(groupData, code, lino, end, terrno);
65,725✔
814

815
  FOREACH(pNode, group) {
142,820!
816
    SScalarParam output = {0};
77,096✔
817

818
    switch (nodeType(pNode)) {
77,096!
819
      case QUERY_NODE_VALUE:
×
820
        break;
×
821
      case QUERY_NODE_COLUMN:
77,096✔
822
      case QUERY_NODE_OPERATOR:
823
      case QUERY_NODE_FUNCTION: {
824
        SExprNode* expNode = (SExprNode*)pNode;
77,096✔
825
        code = createResultData(&expNode->resType, rows, &output);
77,096✔
826
        if (code != TSDB_CODE_SUCCESS) {
77,166!
827
          goto end;
×
828
        }
829
        break;
77,166✔
830
      }
831

832
      default:
×
833
        code = TSDB_CODE_OPS_NOT_SUPPORT;
×
834
        goto end;
×
835
    }
836

837
    if (nodeType(pNode) == QUERY_NODE_COLUMN) {
77,166✔
838
      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;
73,080✔
839
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);
73,080✔
840
      QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);
73,015!
841
      code = colDataAssign(output.columnData, pColInfo, rows, NULL);
73,015✔
842
    } else if (nodeType(pNode) == QUERY_NODE_VALUE) {
4,086!
843
      continue;
×
844
    } else {
845
      code = scalarCalculate(pNode, pBlockList, &output);
4,086✔
846
    }
847

848
    if (code != TSDB_CODE_SUCCESS) {
77,122!
849
      releaseColInfoData(output.columnData);
×
850
      goto end;
×
851
    }
852

853
    void* tmp = taosArrayPush(groupData, &output.columnData);
77,127✔
854
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
77,127!
855
  }
856

857
  int32_t keyLen = 0;
65,724✔
858
  SNode*  node;
859
  FOREACH(node, group) {
142,864✔
860
    SExprNode* pExpr = (SExprNode*)node;
77,140✔
861
    keyLen += pExpr->resType.bytes;
77,140✔
862
  }
863

864
  int32_t nullFlagSize = sizeof(int8_t) * LIST_LENGTH(group);
65,724✔
865
  keyLen += nullFlagSize;
65,724✔
866

867
  keyBuf = taosMemoryCalloc(1, keyLen);
65,724✔
868
  if (keyBuf == NULL) {
65,699!
869
    code = terrno;
×
870
    goto end;
×
871
  }
872

873
  if (initRemainGroups) {
65,699✔
874
    pTableListInfo->remainGroups =
13,018✔
875
        taosHashInit(rows, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
13,047✔
876
    if (pTableListInfo->remainGroups == NULL) {
13,018!
877
      code = terrno;
×
878
      goto end;
×
879
    }
880
  }
881

882
  for (int i = 0; i < rows; i++) {
325,570✔
883
    STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
259,766✔
884
    QUERY_CHECK_NULL(info, code, lino, end, terrno);
259,615!
885

886
    char* isNull = (char*)keyBuf;
259,621✔
887
    char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(group);
259,621!
888
    for (int j = 0; j < taosArrayGetSize(groupData); j++) {
547,217✔
889
      SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);
287,719✔
890

891
      if (colDataIsNull_s(pValue, i)) {
575,220✔
892
        isNull[j] = 1;
577✔
893
      } else {
894
        isNull[j] = 0;
287,033✔
895
        char* data = colDataGetData(pValue, i);
287,033!
896
        if (pValue->info.type == TSDB_DATA_TYPE_JSON) {
287,033✔
897
          if (tTagIsJson(data)) {
350✔
898
            code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
12✔
899
            goto end;
12✔
900
          }
901
          if (tTagIsJsonNull(data)) {
338!
902
            isNull[j] = 1;
×
903
            continue;
×
904
          }
905
          int32_t len = getJsonValueLen(data);
338✔
906
          memcpy(pStart, data, len);
338✔
907
          pStart += len;
338✔
908
        } else if (IS_VAR_DATA_TYPE(pValue->info.type)) {
286,683!
909
          if (varDataTLen(data) > pValue->info.bytes) {
256,115✔
910
            code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
2✔
911
            goto end;
2✔
912
          }
913
          memcpy(pStart, data, varDataTLen(data));
256,113✔
914
          pStart += varDataTLen(data);
256,113✔
915
        } else {
916
          memcpy(pStart, data, pValue->info.bytes);
30,568✔
917
          pStart += pValue->info.bytes;
30,568✔
918
        }
919
      }
920
    }
921

922
    int32_t len = (int32_t)(pStart - (char*)keyBuf);
258,493✔
923
    info->groupId = calcGroupId(keyBuf, len);
258,493✔
924
    if (initRemainGroups) {
259,882✔
925
      // groupId ~ table uid
926
      code = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid),
76,484✔
927
                         sizeof(info->uid));
928
      if (code == TSDB_CODE_DUP_KEY) {
76,502✔
929
        code = TSDB_CODE_SUCCESS;
1,288✔
930
      }
931
      QUERY_CHECK_CODE(code, lino, end);
76,502!
932
    }
933
  }
934

935
  if (tsTagFilterCache) {
65,804!
936
    tableList = taosArrayDup(pTableListInfo->pTableList, NULL);
×
937
    QUERY_CHECK_NULL(tableList, code, lino, end, terrno);
×
938

939
    code = pAPI->metaFn.metaPutTbGroupToCache(pVnode, pTableListInfo->idInfo.suid, context.digest,
×
940
                                              tListLen(context.digest), tableList,
941
                                              taosArrayGetSize(tableList) * sizeof(STableKeyInfo));
×
942
    QUERY_CHECK_CODE(code, lino, end);
×
943
  }
944

945
  //  int64_t st2 = taosGetTimestampUs();
946
  //  qDebug("calculate tag block rows:%d, cost:%ld us", rows, st2-st1);
947

948
end:
65,804✔
949
  taosMemoryFreeClear(keyBuf);
65,818!
950
  taosHashCleanup(ctx.colHash);
65,850✔
951
  taosArrayDestroy(ctx.cInfoList);
65,719✔
952
  blockDataDestroy(pResBlock);
65,728✔
953
  taosArrayDestroy(pBlockList);
65,738✔
954
  taosArrayDestroyEx(pUidTagList, freeItem);
65,720✔
955
  taosArrayDestroyP(groupData, releaseColInfoData);
65,747✔
956
  if (code != TSDB_CODE_SUCCESS) {
65,738✔
957
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
14!
958
  }
959
  return code;
65,733✔
960
}
961

962
static int32_t nameComparFn(const void* p1, const void* p2) {
3,958✔
963
  const char* pName1 = *(const char**)p1;
3,958✔
964
  const char* pName2 = *(const char**)p2;
3,958✔
965

966
  int32_t ret = strcmp(pName1, pName2);
3,958✔
967
  if (ret == 0) {
3,958✔
968
    return 0;
16✔
969
  } else {
970
    return (ret > 0) ? 1 : -1;
3,942✔
971
  }
972
}
973

974
static SArray* getTableNameList(const SNodeListNode* pList) {
463,111✔
975
  int32_t    code = TSDB_CODE_SUCCESS;
463,111✔
976
  int32_t    lino = 0;
463,111✔
977
  int32_t    len = LIST_LENGTH(pList->pNodeList);
463,111!
978
  SListCell* cell = pList->pNodeList->pHead;
463,111✔
979

980
  SArray* pTbList = taosArrayInit(len, POINTER_BYTES);
463,111✔
981
  QUERY_CHECK_NULL(pTbList, code, lino, _end, terrno);
463,118✔
982

983
  for (int i = 0; i < pList->pNodeList->length; i++) {
929,023✔
984
    SValueNode* valueNode = (SValueNode*)cell->pNode;
465,903✔
985
    if (!IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {
465,903!
986
      terrno = TSDB_CODE_INVALID_PARA;
×
987
      taosArrayDestroy(pTbList);
×
988
      return NULL;
×
989
    }
990

991
    char* name = varDataVal(valueNode->datum.p);
465,903✔
992
    void* tmp = taosArrayPush(pTbList, &name);
465,907✔
993
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
465,907!
994
    cell = cell->pNext;
465,907✔
995
  }
996

997
  size_t numOfTables = taosArrayGetSize(pTbList);
463,120✔
998

999
  // order the name
1000
  taosArraySort(pTbList, nameComparFn);
463,114✔
1001

1002
  // remove the duplicates
1003
  SArray* pNewList = taosArrayInit(taosArrayGetSize(pTbList), sizeof(void*));
463,111✔
1004
  QUERY_CHECK_NULL(pNewList, code, lino, _end, terrno);
463,116!
1005
  void*   tmpTbl = taosArrayGet(pTbList, 0);
463,116✔
1006
  QUERY_CHECK_NULL(tmpTbl, code, lino, _end, terrno);
463,108!
1007
  void*   tmp = taosArrayPush(pNewList, tmpTbl);
463,109✔
1008
  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
463,109!
1009

1010
  for (int32_t i = 1; i < numOfTables; ++i) {
465,901✔
1011
    char** name = taosArrayGetLast(pNewList);
2,791✔
1012
    char** nameInOldList = taosArrayGet(pTbList, i);
2,791✔
1013
    QUERY_CHECK_NULL(nameInOldList, code, lino, _end, terrno);
2,790!
1014
    if (strcmp(*name, *nameInOldList) == 0) {
2,790✔
1015
      continue;
16✔
1016
    }
1017

1018
    tmp = taosArrayPush(pNewList, nameInOldList);
2,776✔
1019
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
2,776!
1020
  }
1021

1022
_end:
463,110✔
1023
  taosArrayDestroy(pTbList);
463,110✔
1024
  if (code != TSDB_CODE_SUCCESS) {
463,112!
1025
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1026
    return NULL;
×
1027
  }
1028
  return pNewList;
463,113✔
1029
}
1030

1031
static int tableUidCompare(const void* a, const void* b) {
×
1032
  uint64_t u1 = *(uint64_t*)a;
×
1033
  uint64_t u2 = *(uint64_t*)b;
×
1034

1035
  if (u1 == u2) {
×
1036
    return 0;
×
1037
  }
1038

1039
  return u1 < u2 ? -1 : 1;
×
1040
}
1041

1042
static int32_t filterTableInfoCompare(const void* a, const void* b) {
5,380✔
1043
  STUidTagInfo* p1 = (STUidTagInfo*)a;
5,380✔
1044
  STUidTagInfo* p2 = (STUidTagInfo*)b;
5,380✔
1045

1046
  if (p1->uid == p2->uid) {
5,380!
1047
    return 0;
×
1048
  }
1049

1050
  return p1->uid < p2->uid ? -1 : 1;
5,380!
1051
}
1052

1053
static FilterCondType checkTagCond(SNode* cond) {
1,392,218✔
1054
  if (nodeType(cond) == QUERY_NODE_OPERATOR) {
1,392,218✔
1055
    return FILTER_NO_LOGIC;
382,875✔
1056
  }
1057
  if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
1,009,343!
1058
    return FILTER_AND;
10,083✔
1059
  }
1060
  return FILTER_OTHER;
999,260✔
1061
}
1062

1063
static int32_t optimizeTbnameInCond(void* pVnode, int64_t suid, SArray* list, SNode* cond, SStorageAPI* pAPI) {
1,392,358✔
1064
  int32_t ret = -1;
1,392,358✔
1065
  int32_t ntype = nodeType(cond);
1,392,358✔
1066

1067
  if (ntype == QUERY_NODE_OPERATOR) {
1,392,358✔
1068
    ret = optimizeTbnameInCondImpl(pVnode, list, cond, pAPI, suid);
382,936✔
1069
  }
1070

1071
  if (ntype != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
1,393,929✔
1072
    return ret;
394,598✔
1073
  }
1074

1075
  bool                 hasTbnameCond = false;
999,331✔
1076
  SLogicConditionNode* pNode = (SLogicConditionNode*)cond;
999,331✔
1077
  SNodeList*           pList = (SNodeList*)pNode->pParameterList;
999,331✔
1078

1079
  int32_t len = LIST_LENGTH(pList);
999,331✔
1080
  if (len <= 0) {
999,331!
1081
    return ret;
×
1082
  }
1083

1084
  SListCell* cell = pList->pHead;
999,331✔
1085
  for (int i = 0; i < len; i++) {
3,211,658✔
1086
    if (cell == NULL) break;
2,486,620!
1087
    if (optimizeTbnameInCondImpl(pVnode, list, cell->pNode, pAPI, suid) == 0) {
2,486,620✔
1088
      hasTbnameCond = true;
273,961✔
1089
      break;
273,961✔
1090
    }
1091
    cell = cell->pNext;
2,212,327✔
1092
  }
1093

1094
  taosArraySort(list, filterTableInfoCompare);
998,999✔
1095
  taosArrayRemoveDuplicate(list, filterTableInfoCompare, NULL);
996,540✔
1096

1097
  if (hasTbnameCond) {
997,115✔
1098
    ret = pAPI->metaFn.getTableTagsByUid(pVnode, suid, list);
273,955✔
1099
  }
1100

1101
  return ret;
997,265✔
1102
}
1103

1104
// only return uid that does not contained in pExistedUidList
1105
static int32_t optimizeTbnameInCondImpl(void* pVnode, SArray* pExistedUidList, SNode* pTagCond,
2,867,960✔
1106
                                        SStorageAPI* pStoreAPI, uint64_t suid) {
1107
  if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {
2,867,960✔
1108
    return -1;
128,783✔
1109
  }
1110

1111
  SOperatorNode* pNode = (SOperatorNode*)pTagCond;
2,739,177✔
1112
  if (pNode->opType != OP_TYPE_IN) {
2,739,177✔
1113
    return -1;
1,901,875✔
1114
  }
1115

1116
  if ((pNode->pLeft != NULL && nodeType(pNode->pLeft) == QUERY_NODE_COLUMN &&
837,302!
1117
       ((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME) &&
840,473✔
1118
      (pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {
463,117!
1119
    SNodeListNode* pList = (SNodeListNode*)pNode->pRight;
463,118✔
1120

1121
    int32_t len = LIST_LENGTH(pList->pNodeList);
463,118✔
1122
    if (len <= 0) {
463,118!
1123
      return -1;
×
1124
    }
1125

1126
    SArray*   pTbList = getTableNameList(pList);
463,118✔
1127
    int32_t   numOfTables = taosArrayGetSize(pTbList);
463,113✔
1128
    SHashObj* uHash = NULL;
463,111✔
1129

1130
    size_t numOfExisted = taosArrayGetSize(pExistedUidList);  // len > 0 means there already have uids
463,111✔
1131
    if (numOfExisted > 0) {
463,102!
1132
      uHash = taosHashInit(numOfExisted / 0.7, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
1133
      if (!uHash) {
×
1134
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1135
        return terrno;
×
1136
      }
1137

1138
      for (int i = 0; i < numOfExisted; i++) {
×
1139
        STUidTagInfo* pTInfo = taosArrayGet(pExistedUidList, i);
×
1140
        if (!pTInfo) {
×
1141
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1142
          return terrno;
×
1143
        }
1144
        int32_t       tempRes = taosHashPut(uHash, &pTInfo->uid, sizeof(uint64_t), &i, sizeof(i));
×
1145
        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
×
1146
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
1147
          return tempRes;
×
1148
        }
1149
      }
1150
    }
1151

1152
    for (int i = 0; i < numOfTables; i++) {
928,949✔
1153
      char* name = taosArrayGetP(pTbList, i);
465,862✔
1154

1155
      uint64_t uid = 0, csuid = 0;
465,863✔
1156
      if (pStoreAPI->metaFn.getTableUidByName(pVnode, name, &uid) == 0) {
465,863✔
1157
        ETableType tbType = TSDB_TABLE_MAX;
463,114✔
1158
        if (pStoreAPI->metaFn.getTableTypeSuidByName(pVnode, name, &tbType, &csuid) == 0 && tbType == TSDB_CHILD_TABLE) {
463,114!
1159
          if (suid != csuid) {
463,089✔
1160
            continue;
20✔
1161
          }
1162
          if (NULL == uHash || taosHashGet(uHash, &uid, sizeof(uid)) == NULL) {
463,069!
1163
            STUidTagInfo s = {.uid = uid, .name = name, .pTagVal = NULL};
463,070✔
1164
            void*        tmp = taosArrayPush(pExistedUidList, &s);
463,065✔
1165
            if (!tmp) {
463,065!
1166
              return terrno;
×
1167
            }
1168
          }
1169
        } else {
1170
          taosArrayDestroy(pTbList);
23✔
1171
          taosHashCleanup(uHash);
21✔
1172
          return -1;
21✔
1173
        }
1174
      } else {
1175
        //        qWarn("failed to get tableIds from by table name: %s, reason: %s", name, tstrerror(terrno));
1176
        terrno = 0;
2,765✔
1177
      }
1178
    }
1179

1180
    taosHashCleanup(uHash);
463,087✔
1181
    taosArrayDestroy(pTbList);
463,088✔
1182
    return 0;
463,094✔
1183
  }
1184

1185
  return -1;
374,184✔
1186
}
1187

1188
SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, SArray* pUidTagList, void* pVnode,
1,450,300✔
1189
                                        SStorageAPI* pStorageAPI) {
1190
  int32_t      code = TSDB_CODE_SUCCESS;
1,450,300✔
1191
  int32_t      lino = 0;
1,450,300✔
1192
  SSDataBlock* pResBlock = NULL;
1,450,300✔
1193
  code = createDataBlock(&pResBlock);
1,450,300✔
1194
  QUERY_CHECK_CODE(code, lino, _end);
1,451,305!
1195

1196
  for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {
4,671,107✔
1197
    SColumnInfoData colInfo = {0};
3,217,842✔
1198
    void* tmp = taosArrayGet(pColList, i);
3,217,842✔
1199
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
3,215,729!
1200
    colInfo.info = *(SColumnInfo*)tmp;
3,215,729✔
1201
    code = blockDataAppendColInfo(pResBlock, &colInfo);
3,215,729✔
1202
    QUERY_CHECK_CODE(code, lino, _end);
3,219,802!
1203
  }
1204

1205
  code = blockDataEnsureCapacity(pResBlock, numOfTables);
1,448,448✔
1206
  if (code != TSDB_CODE_SUCCESS) {
1,451,695!
1207
    terrno = code;
×
1208
    blockDataDestroy(pResBlock);
×
1209
    return NULL;
×
1210
  }
1211

1212
  pResBlock->info.rows = numOfTables;
1,451,695✔
1213

1214
  int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);
1,451,695✔
1215

1216
  for (int32_t i = 0; i < numOfTables; i++) {
4,926,149✔
1217
    STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);
3,469,014✔
1218
    QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
3,466,251✔
1219

1220
    for (int32_t j = 0; j < numOfCols; j++) {
10,709,090✔
1221
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, j);
7,233,802✔
1222
      QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
7,225,969!
1223

1224
      if (pColInfo->info.colId == -1) {  // tbname
7,226,836✔
1225
        char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
479,918✔
1226
        if (p1->name != NULL) {
479,918✔
1227
          STR_TO_VARSTR(str, p1->name);
463,068✔
1228
        } else {  // name is not retrieved during filter
1229
          code = pStorageAPI->metaFn.getTableNameByUid(pVnode, p1->uid, str);
16,850✔
1230
          QUERY_CHECK_CODE(code, lino, _end);
16,862!
1231
        }
1232

1233
        code = colDataSetVal(pColInfo, i, str, false);
479,930✔
1234
        QUERY_CHECK_CODE(code, lino, _end);
479,927!
1235
#if TAG_FILTER_DEBUG
1236
        qDebug("tagfilter uid:%ld, tbname:%s", *uid, str + 2);
1237
#endif
1238
      } else {
1239
        STagVal tagVal = {0};
6,746,918✔
1240
        tagVal.cid = pColInfo->info.colId;
6,746,918✔
1241
        if (p1->pTagVal == NULL) {
6,746,918!
1242
          colDataSetNULL(pColInfo, i);
×
1243
        } else {
1244
          const char* p = pStorageAPI->metaFn.extractTagVal(p1->pTagVal, pColInfo->info.type, &tagVal);
6,746,918✔
1245

1246
          if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {
6,776,947✔
1247
            colDataSetNULL(pColInfo, i);
4,076✔
1248
          } else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) {
6,772,871✔
1249
            code = colDataSetVal(pColInfo, i, p, false);
6,480✔
1250
            QUERY_CHECK_CODE(code, lino, _end);
6,481!
1251
          } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
8,535,250!
1252
            char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);
1,771,573✔
1253
            QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
1,768,799!
1254
            varDataSetLen(tmp, tagVal.nData);
1,768,799✔
1255
            memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);
1,768,799✔
1256
            code = colDataSetVal(pColInfo, i, tmp, false);
1,768,799✔
1257
#if TAG_FILTER_DEBUG
1258
            qDebug("tagfilter varch:%s", tmp + 2);
1259
#endif
1260
            taosMemoryFree(tmp);
1,767,111!
1261
            QUERY_CHECK_CODE(code, lino, _end);
1,768,859!
1262
          } else {
1263
            code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false);
4,994,818✔
1264
            QUERY_CHECK_CODE(code, lino, _end);
4,983,780!
1265
#if TAG_FILTER_DEBUG
1266
            if (pColInfo->info.type == TSDB_DATA_TYPE_INT) {
1267
              qDebug("tagfilter int:%d", *(int*)(&tagVal.i64));
1268
            } else if (pColInfo->info.type == TSDB_DATA_TYPE_DOUBLE) {
1269
              qDebug("tagfilter double:%f", *(double*)(&tagVal.i64));
1270
            }
1271
#endif
1272
          }
1273
        }
1274
      }
1275
    }
1276
  }
1277

1278
_end:
1,457,135✔
1279
  if (code != TSDB_CODE_SUCCESS) {
1,457,135!
1280
    blockDataDestroy(pResBlock);
×
1281
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1282
    terrno = code;
×
1283
    return NULL;
×
1284
  }
1285
  return pResBlock;
1,457,135✔
1286
}
1287

1288
static int32_t doSetQualifiedUid(STableListInfo* pListInfo, SArray* pUidList, const SArray* pUidTagList,
1,368,666✔
1289
                                 bool* pResultList, bool addUid) {
1290
  taosArrayClear(pUidList);
1,368,666✔
1291

1292
  STableKeyInfo info = {.uid = 0, .groupId = 0};
1,368,828✔
1293
  int32_t       numOfTables = taosArrayGetSize(pUidTagList);
1,368,828✔
1294
  for (int32_t i = 0; i < numOfTables; ++i) {
4,561,921✔
1295
    if (pResultList[i]) {
3,190,858✔
1296
      STUidTagInfo* tmpTag = (STUidTagInfo*)taosArrayGet(pUidTagList, i);
2,594,751✔
1297
      if (!tmpTag) {
2,592,980!
1298
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1299
        return terrno;
×
1300
      }
1301
      uint64_t uid = tmpTag->uid;
2,592,980✔
1302
      qDebug("tagfilter get uid:%" PRId64 ", res:%d", uid, pResultList[i]);
2,592,980✔
1303

1304
      info.uid = uid;
2,593,010✔
1305
      void* p = taosArrayPush(pListInfo->pTableList, &info);
2,593,010✔
1306
      if (p == NULL) {
2,595,649!
1307
        return terrno;
×
1308
      }
1309

1310
      if (addUid) {
2,595,649!
1311
        void* tmp = taosArrayPush(pUidList, &uid);
×
1312
        if (tmp == NULL) {
×
1313
          return terrno;
×
1314
        }
1315
      }
1316
    }
1317
  }
1318

1319
  return TSDB_CODE_SUCCESS;
1,371,063✔
1320
}
1321

1322
static int32_t copyExistedUids(SArray* pUidTagList, const SArray* pUidList) {
1,392,299✔
1323
  int32_t code = TSDB_CODE_SUCCESS;
1,392,299✔
1324
  int32_t numOfExisted = taosArrayGetSize(pUidList);
1,392,299✔
1325
  if (numOfExisted == 0) {
1,392,721✔
1326
    return code;
1,362,081✔
1327
  }
1328

1329
  for (int32_t i = 0; i < numOfExisted; ++i) {
64,185✔
1330
    uint64_t*    uid = taosArrayGet(pUidList, i);
33,544✔
1331
    if (!uid) {
33,541!
1332
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1333
      return terrno;
×
1334
    }
1335
    STUidTagInfo info = {.uid = *uid};
33,541✔
1336
    void*        tmp = taosArrayPush(pUidTagList, &info);
33,545✔
1337
    if (!tmp) {
33,545!
1338
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1339
      return code;
×
1340
    }
1341
  }
1342
  return code;
30,641✔
1343
}
1344

1345
static int32_t doFilterByTagCond(STableListInfo* pListInfo, SArray* pUidList, SNode* pTagCond, void* pVnode,
4,876,574✔
1346
                                 SIdxFltStatus status, SStorageAPI* pAPI, bool addUid, bool* listAdded) {
1347
  *listAdded = false;
4,876,574✔
1348
  if (pTagCond == NULL) {
4,876,574✔
1349
    return TSDB_CODE_SUCCESS;
3,485,853✔
1350
  }
1351

1352
  terrno = TSDB_CODE_SUCCESS;
1,390,721✔
1353

1354
  int32_t      lino = 0;
1,393,858✔
1355
  int32_t      code = TSDB_CODE_SUCCESS;
1,393,858✔
1356
  SArray*      pBlockList = NULL;
1,393,858✔
1357
  SSDataBlock* pResBlock = NULL;
1,393,858✔
1358
  SScalarParam output = {0};
1,393,858✔
1359
  SArray*      pUidTagList = NULL;
1,393,858✔
1360

1361
  tagFilterAssist ctx = {0};
1,393,858✔
1362
  ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
1,393,858✔
1363
  if (ctx.colHash == NULL) {
1,394,804!
1364
    code = terrno;
×
1365
    QUERY_CHECK_CODE(code, lino, end);
×
1366
  }
1367

1368
  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
1,394,804✔
1369
  if (ctx.cInfoList == NULL) {
1,395,038!
1370
    code = terrno;
×
1371
    QUERY_CHECK_CODE(code, lino, end);
×
1372
  }
1373

1374
  nodesRewriteExprPostOrder(&pTagCond, getColumn, (void*)&ctx);
1,395,038✔
1375
  if (TSDB_CODE_SUCCESS != ctx.code) {
1,394,464!
1376
    terrno = code = ctx.code;
×
1377
    goto end;
×
1378
  }
1379

1380
  SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};
1,394,464✔
1381

1382
  //  int64_t stt = taosGetTimestampUs();
1383
  pUidTagList = taosArrayInit(10, sizeof(STUidTagInfo));
1,394,464✔
1384
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
1,392,943!
1385

1386
  code = copyExistedUids(pUidTagList, pUidList);
1,392,943✔
1387
  QUERY_CHECK_CODE(code, lino, end);
1,393,609!
1388

1389
  FilterCondType condType = checkTagCond(pTagCond);
1,393,609✔
1390

1391
  int32_t filter = optimizeTbnameInCond(pVnode, pListInfo->idInfo.suid, pUidTagList, pTagCond, pAPI);
1,393,964✔
1392
  if (filter == 0) {  // tbname in filter is activated, do nothing and return
1,390,962✔
1393
    taosArrayClear(pUidList);
463,093✔
1394

1395
    int32_t numOfRows = taosArrayGetSize(pUidTagList);
463,091✔
1396
    code = taosArrayEnsureCap(pUidList, numOfRows);
463,089✔
1397
    QUERY_CHECK_CODE(code, lino, end);
463,091!
1398

1399
    for (int32_t i = 0; i < numOfRows; ++i) {
926,158✔
1400
      STUidTagInfo* pInfo = taosArrayGet(pUidTagList, i);
463,066✔
1401
      QUERY_CHECK_NULL(pInfo, code, lino, end, terrno);
463,065!
1402
      void*         tmp = taosArrayPush(pUidList, &pInfo->uid);
463,065✔
1403
      QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
463,067!
1404
    }
1405
    terrno = 0;
463,092✔
1406
  } else {
1407
    if ((condType == FILTER_NO_LOGIC || condType == FILTER_AND) && status != SFLT_NOT_INDEX) {
927,869✔
1408
      code = pAPI->metaFn.getTableTagsByUid(pVnode, pListInfo->idInfo.suid, pUidTagList);
101✔
1409
    } else {
1410
      code = pAPI->metaFn.getTableTags(pVnode, pListInfo->idInfo.suid, pUidTagList);
927,768✔
1411
    }
1412
    if (code != TSDB_CODE_SUCCESS) {
931,695!
1413
      qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), pListInfo->idInfo.suid);
×
1414
      terrno = code;
×
1415
      QUERY_CHECK_CODE(code, lino, end);
×
1416
    }
1417
  }
1418

1419
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
1,394,783✔
1420
  if (numOfTables == 0) {
1,393,754✔
1421
    goto end;
23,744✔
1422
  }
1423

1424
  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);
1,370,010✔
1425
  if (pResBlock == NULL) {
1,370,735!
1426
    code = terrno;
×
1427
    QUERY_CHECK_CODE(code, lino, end);
×
1428
  }
1429

1430
  //  int64_t st1 = taosGetTimestampUs();
1431
  //  qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
1432
  pBlockList = taosArrayInit(2, POINTER_BYTES);
1,370,735✔
1433
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
1,371,500!
1434

1435
  void* tmp = taosArrayPush(pBlockList, &pResBlock);
1,370,990✔
1436
  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
1,370,990!
1437

1438
  code = createResultData(&type, numOfTables, &output);
1,370,990✔
1439
  if (code != TSDB_CODE_SUCCESS) {
1,371,349!
1440
    terrno = code;
×
1441
    QUERY_CHECK_CODE(code, lino, end);
×
1442
  }
1443

1444
  code = scalarCalculate(pTagCond, pBlockList, &output);
1,371,349✔
1445
  if (code != TSDB_CODE_SUCCESS) {
1,370,472✔
1446
    qError("failed to calculate scalar, reason:%s", tstrerror(code));
16!
1447
    terrno = code;
16✔
1448
    QUERY_CHECK_CODE(code, lino, end);
16!
1449
  }
1450

1451
  code = doSetQualifiedUid(pListInfo, pUidList, pUidTagList, (bool*)output.columnData->pData, addUid);
1,370,456✔
1452
  if (code != TSDB_CODE_SUCCESS) {
1,370,764✔
1453
    terrno = code;
492✔
1454
    QUERY_CHECK_CODE(code, lino, end);
×
1455
  }
1456
  *listAdded = true;
1,370,272✔
1457

1458
end:
1,394,032✔
1459
  if (code != TSDB_CODE_SUCCESS) {
1,394,032✔
1460
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
16!
1461
  }
1462
  taosHashCleanup(ctx.colHash);
1,394,032✔
1463
  taosArrayDestroy(ctx.cInfoList);
1,394,787✔
1464
  blockDataDestroy(pResBlock);
1,395,281✔
1465
  taosArrayDestroy(pBlockList);
1,395,355✔
1466
  taosArrayDestroyEx(pUidTagList, freeItem);
1,395,355✔
1467

1468
  colDataDestroy(output.columnData);
1,395,375✔
1469
  taosMemoryFreeClear(output.columnData);
1,395,438!
1470
  return code;
1,395,400✔
1471
}
1472

1473
int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
4,871,543✔
1474
                     STableListInfo* pListInfo, uint8_t* digest, const char* idstr, SStorageAPI* pStorageAPI) {
1475
  int32_t code = TSDB_CODE_SUCCESS;
4,871,543✔
1476
  int32_t lino = 0;
4,871,543✔
1477
  size_t  numOfTables = 0;
4,871,543✔
1478
  bool    listAdded = false;
4,871,543✔
1479

1480
  pListInfo->idInfo.suid = pScanNode->suid;
4,871,543✔
1481
  pListInfo->idInfo.tableType = pScanNode->tableType;
4,871,543✔
1482

1483
  SArray* pUidList = taosArrayInit(8, sizeof(uint64_t));
4,871,543✔
1484
  QUERY_CHECK_NULL(pUidList, code, lino, _error, terrno);
4,882,787✔
1485

1486
  SIdxFltStatus status = SFLT_NOT_INDEX;
4,878,315✔
1487
  if (pScanNode->tableType != TSDB_SUPER_TABLE) {
4,878,315✔
1488
    pListInfo->idInfo.uid = pScanNode->uid;
637,826✔
1489
    if (pStorageAPI->metaFn.isTableExisted(pVnode, pScanNode->uid)) {
637,826✔
1490
      void* tmp = taosArrayPush(pUidList, &pScanNode->uid);
636,087✔
1491
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
636,080!
1492
    }
1493
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, false, &listAdded);
637,834✔
1494
    QUERY_CHECK_CODE(code, lino, _end);
637,822!
1495
  } else {
1496
    T_MD5_CTX context = {0};
4,240,489✔
1497

1498
    if (tsTagFilterCache) {
4,240,489!
1499
      // try to retrieve the result from meta cache
1500
      code = genTagFilterDigest(pTagCond, &context);
×
1501
      QUERY_CHECK_CODE(code, lino, _error);
×
1502

1503
      bool acquired = false;
×
1504
      code = pStorageAPI->metaFn.getCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),
×
1505
                                                    pUidList, &acquired);
1506
      QUERY_CHECK_CODE(code, lino, _error);
×
1507

1508
      if (acquired) {
×
1509
        digest[0] = 1;
×
1510
        memcpy(digest + 1, context.digest, tListLen(context.digest));
×
1511
        qDebug("retrieve table uid list from cache, numOfTables:%d", (int32_t)taosArrayGetSize(pUidList));
×
1512
        goto _end;
×
1513
      }
1514
    }
1515

1516
    if (!pTagCond) {  // no tag filter condition exists, let's fetch all tables of this super table
4,240,489✔
1517
      code = pStorageAPI->metaFn.getChildTableList(pVnode, pScanNode->suid, pUidList);
2,873,040✔
1518
      QUERY_CHECK_CODE(code, lino, _error);
2,882,113!
1519
    } else {
1520
      // failed to find the result in the cache, let try to calculate the results
1521
      if (pTagIndexCond) {
1,367,449✔
1522
        void* pIndex = pStorageAPI->metaFn.getInvertIndex(pVnode);
58,950✔
1523

1524
        SIndexMetaArg metaArg = {.metaEx = pVnode,
118,038✔
1525
                                 .idx = pStorageAPI->metaFn.storeGetIndexInfo(pVnode),
59,043✔
1526
                                 .ivtIdx = pIndex,
1527
                                 .suid = pScanNode->uid};
58,995✔
1528

1529
        status = SFLT_NOT_INDEX;
58,995✔
1530
        code = doFilterTag(pTagIndexCond, &metaArg, pUidList, &status, &pStorageAPI->metaFilter);
58,995✔
1531
        if (code != 0 || status == SFLT_NOT_INDEX) {  // temporarily disable it for performance sake
58,945✔
1532
          qDebug("failed to get tableIds from index, suid:%" PRIu64, pScanNode->uid);
58,764✔
1533
        } else {
1534
          qDebug("succ to get filter result, table num: %d", (int)taosArrayGetSize(pUidList));
181✔
1535
        }
1536
      }
1537
    }
1538

1539
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, tsTagFilterCache, &listAdded);
4,249,475✔
1540
    QUERY_CHECK_CODE(code, lino, _end);
4,240,627✔
1541

1542
    // let's add the filter results into meta-cache
1543
    numOfTables = taosArrayGetSize(pUidList);
4,240,611✔
1544

1545
    if (tsTagFilterCache) {
4,239,246✔
1546
      size_t size = numOfTables * sizeof(uint64_t) + sizeof(int32_t);
75✔
1547
      char*  pPayload = taosMemoryMalloc(size);
75!
1548
      QUERY_CHECK_NULL(pPayload, code, lino, _end, terrno);
×
1549

1550
      *(int32_t*)pPayload = numOfTables;
×
1551
      if (numOfTables > 0) {
×
1552
        void* tmp = taosArrayGet(pUidList, 0);
×
1553
        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
×
1554
        memcpy(pPayload + sizeof(int32_t), tmp, numOfTables * sizeof(uint64_t));
×
1555
      }
1556

1557
      code = pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),
×
1558
                                                    pPayload, size, 1);
1559
      QUERY_CHECK_CODE(code, lino, _error);
×
1560

1561
      digest[0] = 1;
×
1562
      memcpy(digest + 1, context.digest, tListLen(context.digest));
×
1563
    }
1564
  }
1565

1566
_end:
4,877,009✔
1567
  if (!listAdded) {
4,877,009✔
1568
    numOfTables = taosArrayGetSize(pUidList);
3,507,959✔
1569
    for (int i = 0; i < numOfTables; i++) {
13,296,035✔
1570
      void* tmp = taosArrayGet(pUidList, i);
9,783,818✔
1571
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
9,774,732!
1572
      STableKeyInfo info = {.uid = *(uint64_t*)tmp, .groupId = 0};
9,774,732✔
1573

1574
      void* p = taosArrayPush(pListInfo->pTableList, &info);
9,774,732✔
1575
      if (p == NULL) {
9,786,805!
1576
        taosArrayDestroy(pUidList);
×
1577
        return terrno;
×
1578
      }
1579

1580
      qTrace("tagfilter get uid:%" PRIu64 ", %s", info.uid, idstr);
9,786,805✔
1581
    }
1582
  }
1583

1584
_error:
4,881,267✔
1585
  taosArrayDestroy(pUidList);
4,881,267✔
1586
  if (code != TSDB_CODE_SUCCESS) {
4,883,820✔
1587
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
16!
1588
  }
1589
  return code;
4,881,068✔
1590
}
1591

1592
int32_t qGetTableList(int64_t suid, void* pVnode, void* node, SArray** tableList, void* pTaskInfo) {
99✔
1593
  int32_t        code = TSDB_CODE_SUCCESS;
99✔
1594
  int32_t        lino = 0;
99✔
1595
  SSubplan*      pSubplan = (SSubplan*)node;
99✔
1596
  SScanPhysiNode pNode = {0};
99✔
1597
  pNode.suid = suid;
99✔
1598
  pNode.uid = suid;
99✔
1599
  pNode.tableType = TSDB_SUPER_TABLE;
99✔
1600

1601
  STableListInfo* pTableListInfo = tableListCreate();
99✔
1602
  QUERY_CHECK_NULL(pTableListInfo, code, lino, _end, terrno);
99!
1603
  uint8_t         digest[17] = {0};
99✔
1604
  code =
1605
      getTableList(pVnode, &pNode, pSubplan ? pSubplan->pTagCond : NULL, pSubplan ? pSubplan->pTagIndexCond : NULL,
99✔
1606
                   pTableListInfo, digest, "qGetTableList", &((SExecTaskInfo*)pTaskInfo)->storageAPI);
1607
  QUERY_CHECK_CODE(code, lino, _end);
99!
1608
  *tableList = pTableListInfo->pTableList;
99✔
1609
  pTableListInfo->pTableList = NULL;
99✔
1610
  tableListDestroy(pTableListInfo);
99✔
1611

1612
_end:
99✔
1613
  if (code != TSDB_CODE_SUCCESS) {
99!
1614
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1615
  }
1616
  return code;
99✔
1617
}
1618

1619
size_t getTableTagsBufLen(const SNodeList* pGroups) {
9,473✔
1620
  size_t keyLen = 0;
9,473✔
1621

1622
  SNode* node;
1623
  FOREACH(node, pGroups) {
53,055!
1624
    SExprNode* pExpr = (SExprNode*)node;
43,582✔
1625
    keyLen += pExpr->resType.bytes;
43,582✔
1626
  }
1627

1628
  keyLen += sizeof(int8_t) * LIST_LENGTH(pGroups);
9,473!
1629
  return keyLen;
9,473✔
1630
}
1631

1632
int32_t getGroupIdFromTagsVal(void* pVnode, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId,
16✔
1633
                              SStorageAPI* pAPI) {
1634
  SMetaReader mr = {0};
16✔
1635

1636
  pAPI->metaReaderFn.initReader(&mr, pVnode, META_READER_LOCK, &pAPI->metaFn);
16✔
1637
  if (pAPI->metaReaderFn.getEntryGetUidCache(&mr, uid) != 0) {  // table not exist
16!
1638
    pAPI->metaReaderFn.clearReader(&mr);
×
1639
    return TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
1640
  }
1641

1642
  SNodeList* groupNew = NULL;
16✔
1643
  int32_t    code = nodesCloneList(pGroupNode, &groupNew);
16✔
1644
  if (TSDB_CODE_SUCCESS != code) {
16!
1645
    pAPI->metaReaderFn.clearReader(&mr);
×
1646
    return code;
×
1647
  }
1648

1649
  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
16✔
1650
  nodesRewriteExprsPostOrder(groupNew, doTranslateTagExpr, &ctx);
16✔
1651
  if (TSDB_CODE_SUCCESS != ctx.code) {
16!
1652
    nodesDestroyList(groupNew);
×
1653
    pAPI->metaReaderFn.clearReader(&mr);
×
1654
    return code;
×
1655
  }
1656
  char* isNull = (char*)keyBuf;
16✔
1657
  char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(pGroupNode);
16!
1658

1659
  SNode*  pNode;
1660
  int32_t index = 0;
16✔
1661
  FOREACH(pNode, groupNew) {
32!
1662
    SNode*  pNew = NULL;
16✔
1663
    int32_t code = scalarCalculateConstants(pNode, &pNew);
16✔
1664
    if (TSDB_CODE_SUCCESS == code) {
16!
1665
      REPLACE_NODE(pNew);
16✔
1666
    } else {
1667
      nodesDestroyList(groupNew);
×
1668
      pAPI->metaReaderFn.clearReader(&mr);
×
1669
      return code;
×
1670
    }
1671

1672
    if (nodeType(pNew) != QUERY_NODE_VALUE) {
16!
1673
      nodesDestroyList(groupNew);
×
1674
      pAPI->metaReaderFn.clearReader(&mr);
×
1675
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
1676
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
1677
    }
1678
    SValueNode* pValue = (SValueNode*)pNew;
16✔
1679

1680
    if (pValue->node.resType.type == TSDB_DATA_TYPE_NULL || pValue->isNull) {
16!
1681
      isNull[index++] = 1;
×
1682
      continue;
×
1683
    } else {
1684
      isNull[index++] = 0;
16✔
1685
      char* data = nodesGetValueFromNode(pValue);
16✔
1686
      if (pValue->node.resType.type == TSDB_DATA_TYPE_JSON) {
16!
1687
        if (tTagIsJson(data)) {
×
1688
          terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
×
1689
          nodesDestroyList(groupNew);
×
1690
          pAPI->metaReaderFn.clearReader(&mr);
×
1691
          return terrno;
×
1692
        }
1693
        int32_t len = getJsonValueLen(data);
×
1694
        memcpy(pStart, data, len);
×
1695
        pStart += len;
×
1696
      } else if (IS_VAR_DATA_TYPE(pValue->node.resType.type)) {
16!
1697
        memcpy(pStart, data, varDataTLen(data));
4✔
1698
        pStart += varDataTLen(data);
4✔
1699
      } else {
1700
        memcpy(pStart, data, pValue->node.resType.bytes);
12✔
1701
        pStart += pValue->node.resType.bytes;
12✔
1702
      }
1703
    }
1704
  }
1705

1706
  int32_t len = (int32_t)(pStart - (char*)keyBuf);
16✔
1707
  *pGroupId = calcGroupId(keyBuf, len);
16✔
1708

1709
  nodesDestroyList(groupNew);
16✔
1710
  pAPI->metaReaderFn.clearReader(&mr);
16✔
1711
  
1712
  return TSDB_CODE_SUCCESS;
16✔
1713
}
1714

1715
SArray* makeColumnArrayFromList(SNodeList* pNodeList) {
22,355✔
1716
  if (!pNodeList) {
22,355!
1717
    return NULL;
×
1718
  }
1719

1720
  size_t  numOfCols = LIST_LENGTH(pNodeList);
22,355!
1721
  SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
22,355✔
1722
  if (pList == NULL) {
22,358!
1723
    return NULL;
×
1724
  }
1725

1726
  for (int32_t i = 0; i < numOfCols; ++i) {
61,060✔
1727
    SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i);
38,698✔
1728
    if (!pColNode) {
38,690!
1729
      taosArrayDestroy(pList);
×
1730
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
1731
      return NULL;
×
1732
    }
1733

1734
    // todo extract method
1735
    SColumn c = {0};
38,690✔
1736
    c.slotId = pColNode->slotId;
38,690✔
1737
    c.colId = pColNode->colId;
38,690✔
1738
    c.type = pColNode->node.resType.type;
38,690✔
1739
    c.bytes = pColNode->node.resType.bytes;
38,690✔
1740
    c.precision = pColNode->node.resType.precision;
38,690✔
1741
    c.scale = pColNode->node.resType.scale;
38,690✔
1742

1743
    void* tmp = taosArrayPush(pList, &c);
38,702✔
1744
    if (!tmp) {
38,702!
1745
      taosArrayDestroy(pList);
×
1746
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1747
      return NULL;
×
1748
    }
1749
  }
1750

1751
  return pList;
22,362✔
1752
}
1753

1754
int32_t extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
9,011,496✔
1755
                            int32_t type, SColMatchInfo* pMatchInfo) {
1756
  size_t  numOfCols = LIST_LENGTH(pNodeList);
9,011,496!
1757
  int32_t code = TSDB_CODE_SUCCESS;
9,011,496✔
1758
  int32_t lino = 0;
9,011,496✔
1759

1760
  pMatchInfo->matchType = type;
9,011,496✔
1761

1762
  SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchItem));
9,011,496✔
1763
  if (pList == NULL) {
9,013,819!
1764
    code = terrno;
×
1765
    return code;
×
1766
  }
1767

1768
  for (int32_t i = 0; i < numOfCols; ++i) {
46,936,434✔
1769
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);
37,907,867✔
1770
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
37,918,495!
1771
    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {
37,923,810✔
1772
      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
37,828,916✔
1773

1774
      SColMatchItem c = {.needOutput = true};
37,828,916✔
1775
      c.colId = pColNode->colId;
37,828,916✔
1776
      c.srcSlotId = pColNode->slotId;
37,828,916✔
1777
      c.dstSlotId = pNode->slotId;
37,828,916✔
1778
      c.isPk = pColNode->isPk;
37,828,916✔
1779
      c.dataType = pColNode->node.resType;
37,828,916✔
1780
      void* tmp = taosArrayPush(pList, &c);
37,827,294✔
1781
      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
37,827,294!
1782
    }
1783
  }
1784

1785
  // set the output flag for each column in SColMatchInfo, according to the
1786
  *numOfOutputCols = 0;
9,028,567✔
1787
  int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);
9,028,567✔
1788
  for (int32_t i = 0; i < num; ++i) {
54,622,290✔
1789
    SSlotDescNode* pNode = (SSlotDescNode*)nodesListGetNode(pOutputNodeList->pSlots, i);
45,574,420✔
1790
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
45,699,574!
1791

1792
    // todo: add reserve flag check
1793
    // it is a column reserved for the arithmetic expression calculation
1794
    if (pNode->slotId >= numOfCols) {
45,715,612✔
1795
      (*numOfOutputCols) += 1;
7,798,513✔
1796
      continue;
7,798,513✔
1797
    }
1798

1799
    SColMatchItem* info = NULL;
37,917,099✔
1800
    for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {
888,697,718✔
1801
      info = taosArrayGet(pList, j);
888,600,159✔
1802
      QUERY_CHECK_NULL(info, code, lino, _end, terrno);
888,881,826✔
1803
      if (info->dstSlotId == pNode->slotId) {
888,294,362✔
1804
        break;
37,513,743✔
1805
      }
1806
    }
1807

1808
    if (pNode->output) {
37,795,210✔
1809
      (*numOfOutputCols) += 1;
37,718,364✔
1810
    } else if (info != NULL) {
76,846!
1811
      // select distinct tbname from stb where tbname='abc';
1812
      info->needOutput = false;
76,847✔
1813
    }
1814
  }
1815

1816
  pMatchInfo->pList = pList;
9,047,870✔
1817

1818
_end:
9,047,870✔
1819
  if (code != TSDB_CODE_SUCCESS) {
9,047,870!
1820
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1821
  }
1822
  return code;
9,011,933✔
1823
}
1824

1825
static SResSchema createResSchema(int32_t type, int32_t bytes, int32_t slotId, int32_t scale, int32_t precision,
29,712,931✔
1826
                                  const char* name) {
1827
  SResSchema s = {0};
29,712,931✔
1828
  s.scale = scale;
29,712,931✔
1829
  s.type = type;
29,712,931✔
1830
  s.bytes = bytes;
29,712,931✔
1831
  s.slotId = slotId;
29,712,931✔
1832
  s.precision = precision;
29,712,931✔
1833
  tstrncpy(s.name, name, tListLen(s.name));
29,712,931✔
1834

1835
  return s;
29,712,931✔
1836
}
1837

1838
static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType, EColumnType colType) {
28,374,215✔
1839
  SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn));
28,374,215!
1840
  if (pCol == NULL) {
28,374,262!
1841
    return NULL;
×
1842
  }
1843

1844
  pCol->slotId = slotId;
28,374,262✔
1845
  pCol->colId = colId;
28,374,262✔
1846
  pCol->bytes = pType->bytes;
28,374,262✔
1847
  pCol->type = pType->type;
28,374,262✔
1848
  pCol->scale = pType->scale;
28,374,262✔
1849
  pCol->precision = pType->precision;
28,374,262✔
1850
  pCol->dataBlockId = blockId;
28,374,262✔
1851
  pCol->colType = colType;
28,374,262✔
1852
  return pCol;
28,374,262✔
1853
}
1854

1855
int32_t createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) {
29,712,366✔
1856
  int32_t code = TSDB_CODE_SUCCESS;
29,712,366✔
1857
  int32_t lino = 0;
29,712,366✔
1858
  pExp->base.numOfParams = 0;
29,712,366✔
1859
  pExp->base.pParam = NULL;
29,712,366✔
1860
  pExp->pExpr = taosMemoryCalloc(1, sizeof(tExprNode));
29,712,366!
1861
  QUERY_CHECK_NULL(pExp->pExpr, code, lino, _end, terrno);
29,725,307!
1862

1863
  pExp->pExpr->_function.num = 1;
29,725,307✔
1864
  pExp->pExpr->_function.functionId = -1;
29,725,307✔
1865

1866
  int32_t type = nodeType(pNode);
29,725,307✔
1867
  // it is a project query, or group by column
1868
  if (type == QUERY_NODE_COLUMN) {
29,725,307✔
1869
    pExp->pExpr->nodeType = QUERY_NODE_COLUMN;
18,367,912✔
1870
    SColumnNode* pColNode = (SColumnNode*)pNode;
18,367,912✔
1871

1872
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
18,367,912!
1873
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
18,366,963!
1874

1875
    pExp->base.numOfParams = 1;
18,366,963✔
1876

1877
    SDataType* pType = &pColNode->node.resType;
18,366,963✔
1878
    pExp->base.resSchema =
1879
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pColNode->colName);
18,366,963✔
1880

1881
    pExp->base.pParam[0].pCol =
36,730,209✔
1882
        createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType, pColNode->colType);
18,368,555✔
1883
    QUERY_CHECK_NULL(pExp->base.pParam[0].pCol, code, lino, _end, terrno);
18,361,654!
1884

1885
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN;
18,361,654✔
1886
  } else if (type == QUERY_NODE_VALUE) {
11,357,395✔
1887
    pExp->pExpr->nodeType = QUERY_NODE_VALUE;
278,240✔
1888
    SValueNode* pValNode = (SValueNode*)pNode;
278,240✔
1889

1890
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
278,240!
1891
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
278,348!
1892

1893
    pExp->base.numOfParams = 1;
278,348✔
1894

1895
    SDataType* pType = &pValNode->node.resType;
278,348✔
1896
    pExp->base.resSchema =
1897
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pValNode->node.aliasName);
278,348✔
1898
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;
278,290✔
1899
    code = nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param);
278,290✔
1900
    QUERY_CHECK_CODE(code, lino, _end);
257,837!
1901
  } else if (type == QUERY_NODE_FUNCTION) {
11,079,155✔
1902
    pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;
10,840,646✔
1903
    SFunctionNode* pFuncNode = (SFunctionNode*)pNode;
10,840,646✔
1904

1905
    SDataType* pType = &pFuncNode->node.resType;
10,840,646✔
1906
    pExp->base.resSchema =
1907
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pFuncNode->node.aliasName);
10,840,646✔
1908
    tExprNode* pExprNode = pExp->pExpr;
10,840,077✔
1909

1910
    pExprNode->_function.functionId = pFuncNode->funcId;
10,840,077✔
1911
    pExprNode->_function.pFunctNode = pFuncNode;
10,840,077✔
1912
    pExprNode->_function.functionType = pFuncNode->funcType;
10,840,077✔
1913

1914
    tstrncpy(pExprNode->_function.functionName, pFuncNode->functionName, tListLen(pExprNode->_function.functionName));
10,840,077✔
1915

1916
#if 1
1917
    // todo refactor: add the parameter for tbname function
1918
    const char* name = "tbname";
10,840,077✔
1919
    int32_t     len = strlen(name);
10,840,077✔
1920

1921
    if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) &&
10,840,077✔
1922
        pExprNode->_function.functionName[len] == 0) {
701,909!
1923
      pFuncNode->pParameterList = NULL;
701,998✔
1924
      int32_t     code = nodesMakeList(&pFuncNode->pParameterList);
701,998✔
1925
      SValueNode* res = NULL;
702,011✔
1926
      if (TSDB_CODE_SUCCESS == code) {
702,011!
1927
        code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
702,021✔
1928
      }
1929
      QUERY_CHECK_CODE(code, lino, _end);
701,915!
1930
      res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
701,915✔
1931
      code = nodesListAppend(pFuncNode->pParameterList, (SNode*)res);
701,915✔
1932
      if (code != TSDB_CODE_SUCCESS) {
701,893!
1933
        nodesDestroyNode((SNode*)res);
×
1934
        res = NULL;
×
1935
      }
1936
      QUERY_CHECK_CODE(code, lino, _end);
701,900!
1937
    }
1938
#endif
1939

1940
    int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);
10,839,979✔
1941

1942
    pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam));
10,839,979!
1943
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
10,845,584!
1944
    pExp->base.numOfParams = numOfParam;
10,845,584✔
1945

1946
    for (int32_t j = 0; j < numOfParam && TSDB_CODE_SUCCESS == code; ++j) {
23,564,868!
1947
      SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j);
12,717,054✔
1948
      QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
12,718,418!
1949
      if (p1->type == QUERY_NODE_COLUMN) {
12,721,148✔
1950
        SColumnNode* pcn = (SColumnNode*)p1;
10,014,094✔
1951

1952
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN;
10,014,094✔
1953
        pExp->base.pParam[j].pCol =
20,029,474✔
1954
            createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType, pcn->colType);
10,014,094✔
1955
        QUERY_CHECK_NULL(pExp->base.pParam[j].pCol, code, lino, _end, terrno);
10,015,380✔
1956
      } else if (p1->type == QUERY_NODE_VALUE) {
2,707,054✔
1957
        SValueNode* pvn = (SValueNode*)p1;
2,266,033✔
1958
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE;
2,266,033✔
1959
        code = nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param);
2,266,033✔
1960
        QUERY_CHECK_CODE(code, lino, _end);
2,265,285!
1961
      }
1962
    }
1963
    pExp->pExpr->_function.bindExprID = ((SExprNode*)pNode)->bindExprID;
10,847,814✔
1964
  } else if (type == QUERY_NODE_OPERATOR) {
238,509✔
1965
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
231,007✔
1966
    SOperatorNode* pOpNode = (SOperatorNode*)pNode;
231,007✔
1967

1968
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
231,007✔
1969
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
231,041!
1970
    pExp->base.numOfParams = 1;
231,041✔
1971

1972
    SDataType* pType = &pOpNode->node.resType;
231,041✔
1973
    pExp->base.resSchema =
1974
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pOpNode->node.aliasName);
231,041✔
1975
    pExp->pExpr->_optrRoot.pRootNode = pNode;
230,952✔
1976
  } else if (type == QUERY_NODE_CASE_WHEN) {
7,502✔
1977
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
5,534✔
1978
    SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;
5,534✔
1979

1980
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
5,534!
1981
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
5,535!
1982
    pExp->base.numOfParams = 1;
5,535✔
1983

1984
    SDataType* pType = &pCaseNode->node.resType;
5,535✔
1985
    pExp->base.resSchema =
1986
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCaseNode->node.aliasName);
5,535✔
1987
    pExp->pExpr->_optrRoot.pRootNode = pNode;
5,535✔
1988
  } else if (type == QUERY_NODE_LOGIC_CONDITION) {
1,968✔
1989
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
10✔
1990
    SLogicConditionNode* pCond = (SLogicConditionNode*)pNode;
10✔
1991
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
10!
1992
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
10!
1993
    pExp->base.numOfParams = 1;
10✔
1994
    SDataType* pType = &pCond->node.resType;
10✔
1995
    pExp->base.resSchema = createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCond->node.aliasName);
10✔
1996
    pExp->pExpr->_optrRoot.pRootNode = pNode;
10✔
1997
  } else {
1998
    code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
1,958✔
1999
    QUERY_CHECK_CODE(code, lino, _end);
1,958!
2000
  }
2001
  pExp->pExpr->relatedTo = ((SExprNode*)pNode)->relatedTo;
29,705,760✔
2002
_end:
29,705,760✔
2003
  if (code != TSDB_CODE_SUCCESS) {
29,705,760!
2004
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2005
  }
2006
  return code;
29,708,037✔
2007
}
2008

2009
int32_t createExprFromTargetNode(SExprInfo* pExp, STargetNode* pTargetNode) {
29,688,121✔
2010
  return createExprFromOneNode(pExp, pTargetNode->pExpr, pTargetNode->slotId);
29,688,121✔
2011
}
2012

2013
SExprInfo* createExpr(SNodeList* pNodeList, int32_t* numOfExprs) {
2,293✔
2014
  *numOfExprs = LIST_LENGTH(pNodeList);
2,293!
2015
  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
2,293!
2016
  if (!pExprs) {
2,294!
2017
    return NULL;
×
2018
  }
2019

2020
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
17,026✔
2021
    SExprInfo* pExp = &pExprs[i];
14,731✔
2022
    int32_t    code = createExprFromOneNode(pExp, nodesListGetNode(pNodeList, i), i + UD_TAG_COLUMN_INDEX);
14,731✔
2023
    if (code != TSDB_CODE_SUCCESS) {
14,732!
2024
      taosMemoryFreeClear(pExprs);
×
2025
      terrno = code;
×
2026
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2027
      return NULL;
×
2028
    }
2029
  }
2030

2031
  return pExprs;
2,295✔
2032
}
2033

2034
int32_t createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, SExprInfo** pExprInfo, int32_t* numOfExprs) {
11,657,876✔
2035
  QRY_PARAM_CHECK(pExprInfo);
11,657,876!
2036

2037
  int32_t code = 0;
11,657,876✔
2038
  int32_t numOfFuncs = LIST_LENGTH(pNodeList);
11,657,876✔
2039
  int32_t numOfGroupKeys = 0;
11,657,876✔
2040
  if (pGroupKeys != NULL) {
11,657,876✔
2041
    numOfGroupKeys = LIST_LENGTH(pGroupKeys);
397,892!
2042
  }
2043

2044
  *numOfExprs = numOfFuncs + numOfGroupKeys;
11,657,876✔
2045
  if (*numOfExprs == 0) {
11,657,876✔
2046
    return code;
1,624,352✔
2047
  }
2048

2049
  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
10,033,524!
2050
  if (pExprs == NULL) {
10,037,263!
2051
    return terrno;
×
2052
  }
2053

2054
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
39,332,010✔
2055
    STargetNode* pTargetNode = NULL;
29,292,639✔
2056
    if (i < numOfFuncs) {
29,292,639✔
2057
      pTargetNode = (STargetNode*)nodesListGetNode(pNodeList, i);
28,651,678✔
2058
    } else {
2059
      pTargetNode = (STargetNode*)nodesListGetNode(pGroupKeys, i - numOfFuncs);
640,961✔
2060
    }
2061
    if (!pTargetNode) {
29,304,400!
2062
      destroyExprInfo(pExprs, *numOfExprs);
×
2063
      taosMemoryFreeClear(pExprs);
×
2064
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2065
      return terrno;
×
2066
    }
2067

2068
    SExprInfo* pExp = &pExprs[i];
29,304,400✔
2069
    code = createExprFromTargetNode(pExp, pTargetNode);
29,304,400✔
2070
    if (code != TSDB_CODE_SUCCESS) {
29,297,165✔
2071
      destroyExprInfo(pExprs, *numOfExprs);
2,418✔
2072
      taosMemoryFreeClear(pExprs);
×
2073
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2074
      return code;
×
2075
    }
2076
  }
2077

2078
  *pExprInfo = pExprs;
10,039,371✔
2079
  return code;
10,039,371✔
2080
}
2081

2082
static void deleteSubsidiareCtx(void* pData) {
×
2083
  SSubsidiaryResInfo* pCtx = (SSubsidiaryResInfo*)pData;
×
2084
  if (pCtx->pCtx) {
×
2085
    taosMemoryFreeClear(pCtx->pCtx);
×
2086
  }
2087
}
×
2088

2089
// set the output buffer for the selectivity + tag query
2090
static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
11,327,777✔
2091
  int32_t num = 0;
11,327,777✔
2092
  int32_t code = TSDB_CODE_SUCCESS;
11,327,777✔
2093
  int32_t lino = 0;
11,327,777✔
2094

2095
  SArray* pValCtxArray = NULL;
11,327,777✔
2096
  for (int32_t i = numOfOutput - 1; i > 0; --i) {  // select Func is at the end of the list
30,466,917✔
2097
    int32_t funcIdx = pCtx[i].pExpr->pExpr->_function.bindExprID;
19,139,140✔
2098
    if (funcIdx > 0) {
19,139,140✔
2099
      if (pValCtxArray == NULL) {
3,493✔
2100
        // the end of the list is the select function of biggest index
2101
        pValCtxArray = taosArrayInit_s(sizeof(SSubsidiaryResInfo*), funcIdx);
2,479✔
2102
        if (pValCtxArray == NULL) {
2,480✔
2103
          return terrno;
1✔
2104
        }
2105
      }
2106
      if (funcIdx > pValCtxArray->size) {
3,493!
2107
        qError("funcIdx:%d is out of range", funcIdx);
×
2108
        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2109
        return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2110
      }
2111
      SSubsidiaryResInfo* pSubsidiary = &pCtx[i].subsidiaries;
3,493✔
2112
      pSubsidiary->pCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
3,493!
2113
      if (pSubsidiary->pCtx == NULL) {
3,495!
2114
        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2115
        return terrno;
×
2116
      }
2117
      pSubsidiary->num = 0;
3,495✔
2118
      taosArraySet(pValCtxArray, funcIdx - 1, &pSubsidiary);
3,495✔
2119
    }
2120
  }
2121

2122
  SqlFunctionCtx*  p = NULL;
11,327,777✔
2123
  SqlFunctionCtx** pValCtx = NULL;
11,327,777✔
2124
  if (pValCtxArray == NULL) {
11,327,777✔
2125
    pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
11,324,727!
2126
    if (pValCtx == NULL) {
11,331,696!
2127
      QUERY_CHECK_CODE(terrno, lino, _end);
×
2128
    }
2129
  }
2130

2131
  for (int32_t i = 0; i < numOfOutput; ++i) {
40,859,003✔
2132
    const char* pName = pCtx[i].pExpr->pExpr->_function.functionName;
29,521,898✔
2133
    if ((strcmp(pName, "_select_value") == 0)) {
29,521,898✔
2134
      if (pValCtxArray == NULL) {
1,174,155✔
2135
        pValCtx[num++] = &pCtx[i];
1,170,108✔
2136
      } else {
2137
        int32_t bindFuncIndex = pCtx[i].pExpr->pExpr->relatedTo;  // start from index 1;
4,047✔
2138
        if (bindFuncIndex > 0) {  // 0 is default index related to the select function
4,047!
2139
          bindFuncIndex -= 1;
4,707✔
2140
        }
2141
        SSubsidiaryResInfo** pSubsidiary = taosArrayGet(pValCtxArray, bindFuncIndex);
4,047✔
2142
        if(pSubsidiary == NULL) {
4,829!
2143
          QUERY_CHECK_CODE(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, lino, _end);
×
2144
        }
2145
        (*pSubsidiary)->pCtx[(*pSubsidiary)->num] = &pCtx[i];
4,829✔
2146
        (*pSubsidiary)->num++;
4,829✔
2147
      }
2148
    } else if (fmIsSelectFunc(pCtx[i].functionId)) {
28,347,743✔
2149
       if (pValCtxArray == NULL) {
2,572,038✔
2150
        p = &pCtx[i];
2,568,207✔
2151
       }
2152
    }
2153
  }
2154

2155
  if (p != NULL) {
11,337,105✔
2156
    p->subsidiaries.pCtx = pValCtx;
2,168,292✔
2157
    p->subsidiaries.num = num;
2,168,292✔
2158
  } else {
2159
    taosMemoryFreeClear(pValCtx);
9,168,813!
2160
  }
2161

2162
_end:
8,893✔
2163
  if (code != TSDB_CODE_SUCCESS) {
11,338,686!
2164
    taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2165
    taosMemoryFreeClear(pValCtx);
×
2166
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2167
  } else {
2168
    taosArrayDestroy(pValCtxArray);
11,338,686✔
2169
  }
2170
  return code;
11,317,627✔
2171
}
2172

2173
SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowEntryInfoOffset,
11,322,441✔
2174
                                     SFunctionStateStore* pStore) {
2175
  int32_t         code = TSDB_CODE_SUCCESS;
11,322,441✔
2176
  int32_t         lino = 0;
11,322,441✔
2177
  SqlFunctionCtx* pFuncCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));
11,322,441!
2178
  if (pFuncCtx == NULL) {
11,322,835!
2179
    return NULL;
×
2180
  }
2181

2182
  *rowEntryInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t));
11,322,835!
2183
  if (*rowEntryInfoOffset == 0) {
11,335,179!
2184
    taosMemoryFreeClear(pFuncCtx);
×
2185
    return NULL;
×
2186
  }
2187

2188
  for (int32_t i = 0; i < numOfOutput; ++i) {
40,885,418✔
2189
    SExprInfo* pExpr = &pExprInfo[i];
29,540,954✔
2190

2191
    SExprBasicInfo* pFunct = &pExpr->base;
29,540,954✔
2192
    SqlFunctionCtx* pCtx = &pFuncCtx[i];
29,540,954✔
2193

2194
    pCtx->functionId = -1;
29,540,954✔
2195
    pCtx->pExpr = pExpr;
29,540,954✔
2196

2197
    if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {
29,540,954✔
2198
      SFuncExecEnv env = {0};
10,842,332✔
2199
      pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;
10,842,332✔
2200
      pCtx->isPseudoFunc = fmIsWindowPseudoColumnFunc(pCtx->functionId);
10,842,332✔
2201
      pCtx->isNotNullFunc = fmIsNotNullOutputFunc(pCtx->functionId);
10,832,838✔
2202

2203
      bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);
10,837,613✔
2204
      if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId)) {
18,073,943✔
2205
        if (!isUdaf) {
7,246,822✔
2206
          code = fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
7,246,790✔
2207
          QUERY_CHECK_CODE(code, lino, _end);
7,237,776!
2208
        } else {
2209
          char* udfName = pExpr->pExpr->_function.pFunctNode->functionName;
32✔
2210
          pCtx->udfName = taosStrdup(udfName);
32!
2211
          QUERY_CHECK_NULL(pCtx->udfName, code, lino, _end, terrno);
32!
2212

2213
          code = fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet);
32✔
2214
          QUERY_CHECK_CODE(code, lino, _end);
32!
2215
        }
2216
        bool tmp = pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
7,237,808✔
2217
        if (!tmp) {
7,238,766!
2218
          code = terrno;
×
2219
          QUERY_CHECK_CODE(code, lino, _end);
×
2220
        }
2221
      } else {
2222
        code = fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);
3,582,822✔
2223
        if (code != TSDB_CODE_SUCCESS && isUdaf) {
3,583,768!
2224
          code = TSDB_CODE_SUCCESS;
261✔
2225
        }
2226
        QUERY_CHECK_CODE(code, lino, _end);
3,583,768!
2227

2228
        if (pCtx->sfp.getEnv != NULL) {
3,583,768✔
2229
          bool tmp = pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
1,899,609✔
2230
          if (!tmp) {
1,897,849✔
2231
            code = terrno;
1,005✔
2232
            QUERY_CHECK_CODE(code, lino, _end);
×
2233
          }
2234
        }
2235
      }
2236
      pCtx->resDataInfo.interBufSize = env.calcMemSize;
10,822,651✔
2237
    } else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR ||
18,698,622✔
2238
               pExpr->pExpr->nodeType == QUERY_NODE_VALUE) {
278,301!
2239
      // for simple column, the result buffer needs to hold at least one element.
2240
      pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes;
18,701,230✔
2241
    }
2242

2243
    pCtx->input.numOfInputCols = pFunct->numOfParams;
29,521,273✔
2244
    pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
29,521,273✔
2245
    QUERY_CHECK_NULL(pCtx->input.pData, code, lino, _end, terrno);
29,545,839!
2246
    pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
29,545,839!
2247
    QUERY_CHECK_NULL(pCtx->input.pColumnDataAgg, code, lino, _end, terrno);
29,550,239!
2248

2249
    pCtx->pTsOutput = NULL;
29,550,239✔
2250
    pCtx->resDataInfo.bytes = pFunct->resSchema.bytes;
29,550,239✔
2251
    pCtx->resDataInfo.type = pFunct->resSchema.type;
29,550,239✔
2252
    pCtx->order = TSDB_ORDER_ASC;
29,550,239✔
2253
    pCtx->start.key = INT64_MIN;
29,550,239✔
2254
    pCtx->end.key = INT64_MIN;
29,550,239✔
2255
    pCtx->numOfParams = pExpr->base.numOfParams;
29,550,239✔
2256
    pCtx->param = pFunct->pParam;
29,550,239✔
2257
    pCtx->saveHandle.currentPage = -1;
29,550,239✔
2258
    pCtx->pStore = pStore;
29,550,239✔
2259
    pCtx->hasWindowOrGroup = false;
29,550,239✔
2260
    pCtx->needCleanup = false;
29,550,239✔
2261
  }
2262

2263
  for (int32_t i = 1; i < numOfOutput; ++i) {
30,493,763✔
2264
    (*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) +
19,149,299✔
2265
                                         pFuncCtx[i - 1].resDataInfo.interBufSize);
19,149,299✔
2266
  }
2267

2268
  code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);
11,344,464✔
2269
  QUERY_CHECK_CODE(code, lino, _end);
11,320,666!
2270

2271
_end:
11,320,666✔
2272
  if (code != TSDB_CODE_SUCCESS) {
11,320,666!
2273
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2274
    for (int32_t i = 0; i < numOfOutput; ++i) {
×
2275
      taosMemoryFree(pFuncCtx[i].input.pData);
×
2276
      taosMemoryFree(pFuncCtx[i].input.pColumnDataAgg);
×
2277
    }
2278
    taosMemoryFreeClear(*rowEntryInfoOffset);
×
2279
    taosMemoryFreeClear(pFuncCtx);
×
2280

2281
    terrno = code;
×
2282
    return NULL;
×
2283
  }
2284
  return pFuncCtx;
11,320,666✔
2285
}
2286

2287
// NOTE: sources columns are more than the destination SSDatablock columns.
2288
// doFilter in table scan needs every column even its output is false
2289
int32_t relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols, bool outputEveryColumn) {
1,113,040✔
2290
  int32_t code = TSDB_CODE_SUCCESS;
1,113,040✔
2291
  size_t  numOfSrcCols = taosArrayGetSize(pCols);
1,113,040✔
2292

2293
  int32_t i = 0, j = 0;
1,113,266✔
2294
  while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) {
2,546,761✔
2295
    SColumnInfoData* p = taosArrayGet(pCols, i);
1,434,242✔
2296
    if (!p) {
1,433,345!
2297
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2298
      return terrno;
×
2299
    }
2300
    SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);
1,433,345✔
2301
    if (!pmInfo) {
1,433,142!
2302
      return terrno;
×
2303
    }
2304

2305
    if (p->info.colId == pmInfo->colId) {
1,433,271✔
2306
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->dstSlotId);
1,348,588✔
2307
      if (!pDst) {
1,348,303!
2308
        return terrno;
×
2309
      }
2310
      code = colDataAssign(pDst, p, pBlock->info.rows, &pBlock->info);
1,348,303✔
2311
      if (code != TSDB_CODE_SUCCESS) {
1,348,820✔
2312
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
8!
2313
        return code;
×
2314
      }
2315
      i++;
1,348,812✔
2316
      j++;
1,348,812✔
2317
    } else if (p->info.colId < pmInfo->colId) {
84,683!
2318
      i++;
84,683✔
2319
    } else {
2320
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
2321
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2322
    }
2323
  }
2324
  return code;
1,113,122✔
2325
}
2326

2327
SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
2,832,605✔
2328
  SInterval interval = {
2,832,605✔
2329
      .interval = pTableScanNode->interval,
2,832,605✔
2330
      .sliding = pTableScanNode->sliding,
2,832,605✔
2331
      .intervalUnit = pTableScanNode->intervalUnit,
2,832,605✔
2332
      .slidingUnit = pTableScanNode->slidingUnit,
2,832,605✔
2333
      .offset = pTableScanNode->offset,
2,832,605✔
2334
      .precision = pTableScanNode->scan.node.pOutputDataBlockDesc->precision,
2,832,605✔
2335
      .timeRange = pTableScanNode->scanRange,
2336
  };
2337
  calcIntervalAutoOffset(&interval);
2,832,605✔
2338

2339
  return interval;
2,834,056✔
2340
}
2341

2342
SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {
792,460✔
2343
  SColumn c = {0};
792,460✔
2344

2345
  c.slotId = pColNode->slotId;
792,460✔
2346
  c.colId = pColNode->colId;
792,460✔
2347
  c.type = pColNode->node.resType.type;
792,460✔
2348
  c.bytes = pColNode->node.resType.bytes;
792,460✔
2349
  c.scale = pColNode->node.resType.scale;
792,460✔
2350
  c.precision = pColNode->node.resType.precision;
792,460✔
2351
  return c;
792,460✔
2352
}
2353

2354
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode,
4,894,933✔
2355
                               const SReadHandle* readHandle) {
2356
  pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
4,894,933✔
2357
  pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);
4,894,933!
2358

2359
  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
4,894,933!
2360
  if (!pCond->colList) {
4,906,549!
2361
    return terrno;
×
2362
  }
2363
  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
4,906,549✔
2364
  if (pCond->pSlotList == NULL) {
4,899,817!
2365
    taosMemoryFreeClear(pCond->colList);
×
2366
    return terrno;
×
2367
  }
2368

2369
  // TODO: get it from stable scan node
2370
  pCond->twindows = pTableScanNode->scanRange;
4,899,817✔
2371
  pCond->suid = pTableScanNode->scan.suid;
4,899,817✔
2372
  pCond->type = TIMEWINDOW_RANGE_CONTAINED;
4,899,817✔
2373
  pCond->startVersion = -1;
4,899,817✔
2374
  pCond->endVersion = -1;
4,899,817✔
2375
  pCond->skipRollup = readHandle->skipRollup;
4,899,817✔
2376

2377
  // allowed read stt file optimization mode
2378
  pCond->notLoadData = (pTableScanNode->dataRequired == FUNC_DATA_REQUIRED_NOT_LOAD) &&
9,888,999✔
2379
                       (pTableScanNode->scan.node.pConditions == NULL) && (pTableScanNode->interval == 0);
4,899,817✔
2380

2381
  int32_t j = 0;
4,899,817✔
2382
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
29,356,199✔
2383
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pTableScanNode->scan.pScanCols, i);
24,446,799✔
2384
    if (!pNode) {
24,459,172✔
2385
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
2,790!
2386
      return terrno;
2,790✔
2387
    }
2388
    SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
24,456,382✔
2389
    if (pColNode->colType == COLUMN_TYPE_TAG) {
24,456,382!
2390
      continue;
×
2391
    }
2392

2393
    pCond->colList[j].type = pColNode->node.resType.type;
24,456,382✔
2394
    pCond->colList[j].bytes = pColNode->node.resType.bytes;
24,456,382✔
2395
    pCond->colList[j].colId = pColNode->colId;
24,456,382✔
2396
    pCond->colList[j].pk = pColNode->isPk;
24,456,382✔
2397

2398
    pCond->pSlotList[j] = pNode->slotId;
24,456,382✔
2399
    j += 1;
24,456,382✔
2400
  }
2401

2402
  pCond->numOfCols = j;
4,909,400✔
2403
  return TSDB_CODE_SUCCESS;
4,909,400✔
2404
}
2405

2406
int32_t initQueryTableDataCondWithColArray(SQueryTableDataCond* pCond, SQueryTableDataCond* pOrgCond,
×
2407
                                           const SReadHandle* readHandle, SArray* colArray) {
2408
  int32_t code = TSDB_CODE_SUCCESS;
×
2409
  int32_t lino = 0;
×
2410

2411
  pCond->order = TSDB_ORDER_ASC;
×
2412
  pCond->numOfCols = (int32_t)taosArrayGetSize(colArray);
×
2413

2414
  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
×
2415
  QUERY_CHECK_NULL(pCond->colList, code, lino, _return, terrno);
×
2416

2417
  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
×
2418
  QUERY_CHECK_NULL(pCond->pSlotList, code, lino, _return, terrno);
×
2419

2420
  pCond->twindows = pOrgCond->twindows;
×
2421
  pCond->type = pOrgCond->type;
×
2422
  pCond->startVersion = -1;
×
2423
  pCond->endVersion = -1;
×
2424
  pCond->skipRollup = true;
×
2425
  pCond->notLoadData = false;
×
2426

2427
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
×
2428
    SColIdPair* pColPair = taosArrayGet(colArray, i);
×
2429
    QUERY_CHECK_NULL(pColPair, code, lino, _return, terrno);
×
2430

2431
    bool find = false;
×
2432
    for (int32_t j = 0; j < pOrgCond->numOfCols; ++j) {
×
2433
      if (pOrgCond->colList[j].colId == pColPair->vtbColId) {
×
2434
        pCond->colList[i].type = pOrgCond->colList[j].type;
×
2435
        pCond->colList[i].bytes = pOrgCond->colList[j].bytes;
×
2436
        pCond->colList[i].colId = pColPair->orgColId;
×
2437
        pCond->colList[i].pk = pOrgCond->colList[j].pk;
×
2438
        pCond->pSlotList[i] = i;
×
2439
        find = true;
×
2440
        break;
×
2441
      }
2442
    }
2443
    QUERY_CHECK_CONDITION(find, code, lino, _return, TSDB_CODE_NOT_FOUND);
×
2444
  }
2445

2446
  return code;
×
2447
_return:
×
2448
  qError("%s failed at line %d since %s", __func__, lino, tstrerror(terrno));
×
2449
  taosMemoryFreeClear(pCond->colList);
×
2450
  taosMemoryFreeClear(pCond->pSlotList);
×
2451
  return code;
×
2452
}
2453

2454
void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {
11,892,937✔
2455
  taosMemoryFreeClear(pCond->colList);
11,892,937!
2456
  taosMemoryFreeClear(pCond->pSlotList);
11,893,617!
2457
}
11,893,581✔
2458

2459
int32_t convertFillType(int32_t mode) {
564,581✔
2460
  int32_t type = TSDB_FILL_NONE;
564,581✔
2461
  switch (mode) {
564,581!
2462
    case FILL_MODE_PREV:
14,970✔
2463
      type = TSDB_FILL_PREV;
14,970✔
2464
      break;
14,970✔
2465
    case FILL_MODE_NONE:
×
2466
      type = TSDB_FILL_NONE;
×
2467
      break;
×
2468
    case FILL_MODE_NULL:
32,577✔
2469
      type = TSDB_FILL_NULL;
32,577✔
2470
      break;
32,577✔
2471
    case FILL_MODE_NULL_F:
30,094✔
2472
      type = TSDB_FILL_NULL_F;
30,094✔
2473
      break;
30,094✔
2474
    case FILL_MODE_NEXT:
15,142✔
2475
      type = TSDB_FILL_NEXT;
15,142✔
2476
      break;
15,142✔
2477
    case FILL_MODE_VALUE:
157,650✔
2478
      type = TSDB_FILL_SET_VALUE;
157,650✔
2479
      break;
157,650✔
2480
    case FILL_MODE_VALUE_F:
289,771✔
2481
      type = TSDB_FILL_SET_VALUE_F;
289,771✔
2482
      break;
289,771✔
2483
    case FILL_MODE_LINEAR:
14,354✔
2484
      type = TSDB_FILL_LINEAR;
14,354✔
2485
      break;
14,354✔
2486
    case FILL_MODE_NEAR:
10,030✔
2487
      type = TSDB_FILL_NEAR;
10,030✔
2488
      break;
10,030✔
2489
    default:
×
2490
      type = TSDB_FILL_NONE;
×
2491
  }
2492

2493
  return type;
564,581✔
2494
}
2495

2496
void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery) {
31,784,258✔
2497
  if (ascQuery) {
31,784,258✔
2498
    *w = getAlignQueryTimeWindow(pInterval, ts);
31,503,748✔
2499
  } else {
2500
    // the start position of the first time window in the endpoint that spreads beyond the queried last timestamp
2501
    *w = getAlignQueryTimeWindow(pInterval, ts);
280,510✔
2502

2503
    int64_t key = w->skey;
280,505✔
2504
    while (key < ts) {  // moving towards end
864,470✔
2505
      key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);
848,302✔
2506
      if (key > ts) {
848,299✔
2507
        break;
264,334✔
2508
      }
2509

2510
      w->skey = key;
583,965✔
2511
    }
2512
    w->ekey = taosTimeAdd(w->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
280,502✔
2513
  }
2514
}
31,784,045✔
2515

2516
static STimeWindow doCalculateTimeWindow(int64_t ts, SInterval* pInterval) {
934,550✔
2517
  STimeWindow w = {0};
934,550✔
2518

2519
  w.skey = taosTimeTruncate(ts, pInterval);
934,550✔
2520
  w.ekey = taosTimeGetIntervalEnd(w.skey, pInterval);
934,630✔
2521
  return w;
934,567✔
2522
}
2523

2524
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {
374,891✔
2525
  STimeWindow win = *pWindow;
374,891✔
2526
  STimeWindow save = win;
374,891✔
2527
  while (win.skey <= ts && win.ekey >= ts) {
1,623,376✔
2528
    save = win;
1,248,484✔
2529
    // get previous time window
2530
    getNextTimeWindow(pInterval, &win, order == TSDB_ORDER_ASC ? TSDB_ORDER_DESC : TSDB_ORDER_ASC);
1,248,484✔
2531
  }
2532

2533
  return save;
374,892✔
2534
}
2535

2536
// get the correct time window according to the handled timestamp
2537
// todo refactor
2538
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
33,275,557✔
2539
                                int32_t order) {
2540
  STimeWindow w = {0};
33,275,557✔
2541
  if (pResultRowInfo->cur.pageId == -1) {  // the first window, from the previous stored value
33,275,557✔
2542
    getInitialStartTimeWindow(pInterval, ts, &w, (order == TSDB_ORDER_ASC));
31,784,254✔
2543
    return w;
31,784,037✔
2544
  }
2545

2546
  SResultRow* pRow = getResultRowByPos(pBuf, &pResultRowInfo->cur, false);
1,491,303✔
2547
  if (pRow) {
1,491,232!
2548
    TAOS_SET_OBJ_ALIGNED(&w, pRow->win);
1,491,235✔
2549
  }
2550

2551
  // in case of typical time window, we can calculate time window directly.
2552
  if (w.skey > ts || w.ekey < ts) {
1,491,232✔
2553
    w = doCalculateTimeWindow(ts, pInterval);
934,554✔
2554
  }
2555

2556
  if (pInterval->interval != pInterval->sliding) {
1,491,241✔
2557
    // it is an sliding window query, in which sliding value is not equalled to
2558
    // interval value, and we need to find the first qualified time window.
2559
    w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
374,892✔
2560
  }
2561

2562
  return w;
1,491,254✔
2563
}
2564

2565
TSKEY getNextTimeWindowStart(const SInterval* pInterval, TSKEY start, int32_t order) {
244,222,319✔
2566
  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
244,222,319✔
2567
  TSKEY   nextStart = taosTimeAdd(start, -1 * pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
244,222,319✔
2568
  nextStart = taosTimeAdd(nextStart, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision, NULL);
244,132,487✔
2569
  nextStart = taosTimeAdd(nextStart, pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
244,317,458✔
2570
  return nextStart;
244,025,961✔
2571
}
2572

2573
void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t order) {
243,182,402✔
2574
  tw->skey = getNextTimeWindowStart(pInterval, tw->skey, order);
243,182,402✔
2575
  tw->ekey = taosTimeAdd(tw->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
243,163,147✔
2576
}
243,213,768✔
2577

2578
bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {
13,485,603✔
2579
  return (pLimitInfo->limit.limit != -1 || pLimitInfo->limit.offset != -1 || pLimitInfo->slimit.limit != -1 ||
26,950,850!
2580
          pLimitInfo->slimit.offset != -1);
13,465,247✔
2581
}
2582

2583
bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo) {
×
2584
  return (pLimitInfo->slimit.limit != -1 || pLimitInfo->slimit.offset != -1);
×
2585
}
2586

2587
void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo) {
14,116,930✔
2588
  SLimit limit = {.limit = getLimit(pLimit), .offset = getOffset(pLimit)};
14,116,930✔
2589
  SLimit slimit = {.limit = getLimit(pSLimit), .offset = getOffset(pSLimit)};
14,121,469✔
2590

2591
  pLimitInfo->limit = limit;
14,123,354✔
2592
  pLimitInfo->slimit = slimit;
14,123,354✔
2593
  pLimitInfo->remainOffset = limit.offset;
14,123,354✔
2594
  pLimitInfo->remainGroupOffset = slimit.offset;
14,123,354✔
2595
}
14,123,354✔
2596

2597
void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo) {
2,521,104✔
2598
  pLimitInfo->numOfOutputRows = 0;
2,521,104✔
2599
  pLimitInfo->remainOffset = pLimitInfo->limit.offset;
2,521,104✔
2600
}
2,521,104✔
2601

2602
int32_t tableListGetSize(const STableListInfo* pTableList, int32_t* pRes) {
17,349,430✔
2603
  if (taosArrayGetSize(pTableList->pTableList) != taosHashGetSize(pTableList->map)) {
17,349,430!
2604
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
×
2605
    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
2606
  }
2607
  (*pRes) = taosArrayGetSize(pTableList->pTableList);
17,360,360✔
2608
  return TSDB_CODE_SUCCESS;
17,358,806✔
2609
}
2610

2611
uint64_t tableListGetSuid(const STableListInfo* pTableList) { return pTableList->idInfo.suid; }
51,227✔
2612

2613
STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index) {
7,354,788✔
2614
  if (taosArrayGetSize(pTableList->pTableList) == 0) {
7,354,788✔
2615
    return NULL;
554✔
2616
  }
2617

2618
  return taosArrayGet(pTableList->pTableList, index);
7,355,128✔
2619
}
2620

2621
int32_t tableListFind(const STableListInfo* pTableList, uint64_t uid, int32_t startIndex) {
181✔
2622
  int32_t numOfTables = taosArrayGetSize(pTableList->pTableList);
181✔
2623
  if (startIndex >= numOfTables) {
181!
2624
    return -1;
×
2625
  }
2626

2627
  for (int32_t i = startIndex; i < numOfTables; ++i) {
3,493✔
2628
    STableKeyInfo* p = taosArrayGet(pTableList->pTableList, i);
3,491✔
2629
    if (!p) {
3,491!
2630
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2631
      return -1;
×
2632
    }
2633
    if (p->uid == uid) {
3,492✔
2634
      return i;
180✔
2635
    }
2636
  }
2637
  return -1;
2✔
2638
}
2639

2640
void tableListGetSourceTableInfo(const STableListInfo* pTableList, uint64_t* psuid, uint64_t* uid, int32_t* type) {
16,944✔
2641
  *psuid = pTableList->idInfo.suid;
16,944✔
2642
  *uid = pTableList->idInfo.uid;
16,944✔
2643
  *type = pTableList->idInfo.tableType;
16,944✔
2644
}
16,944✔
2645

2646
uint64_t tableListGetTableGroupId(const STableListInfo* pTableList, uint64_t tableUid) {
16,422,118✔
2647
  int32_t* slot = taosHashGet(pTableList->map, &tableUid, sizeof(tableUid));
16,422,118✔
2648
  if (slot == NULL) {
16,426,668!
2649
    return -1;
×
2650
  }
2651

2652
  STableKeyInfo* pKeyInfo = taosArrayGet(pTableList->pTableList, *slot);
16,426,668✔
2653

2654
  return pKeyInfo->groupId;
16,424,280✔
2655
}
2656

2657
// TODO handle the group offset info, fix it, the rule of group output will be broken by this function
2658
int32_t tableListAddTableInfo(STableListInfo* pTableList, uint64_t uid, uint64_t gid) {
10,527✔
2659
  int32_t code = TSDB_CODE_SUCCESS;
10,527✔
2660
  int32_t lino = 0;
10,527✔
2661
  if (pTableList->map == NULL) {
10,527!
2662
    pTableList->map = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
×
2663
    QUERY_CHECK_NULL(pTableList->map, code, lino, _end, terrno);
×
2664
  }
2665

2666
  STableKeyInfo keyInfo = {.uid = uid, .groupId = gid};
10,527✔
2667
  void*         p = taosHashGet(pTableList->map, &uid, sizeof(uid));
10,527✔
2668
  if (p != NULL) {
10,535✔
2669
    qInfo("table:%" PRId64 " already in tableIdList, ignore it", uid);
3!
2670
    goto _end;
3✔
2671
  }
2672

2673
  void* tmp = taosArrayPush(pTableList->pTableList, &keyInfo);
10,532✔
2674
  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
10,525!
2675

2676
  int32_t slot = (int32_t)taosArrayGetSize(pTableList->pTableList) - 1;
10,525✔
2677
  code = taosHashPut(pTableList->map, &uid, sizeof(uid), &slot, sizeof(slot));
10,521✔
2678
  if (code != TSDB_CODE_SUCCESS) {
10,540!
2679
    // we have checked the existence of uid in hash map above
2680
    QUERY_CHECK_CONDITION((code != TSDB_CODE_DUP_KEY), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
×
2681
    taosArrayPopTailBatch(pTableList->pTableList, 1);  // let's pop the last element in the array list
×
2682
  }
2683

2684
_end:
10,540✔
2685
  if (code != TSDB_CODE_SUCCESS) {
10,540!
2686
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
2687
  } else {
2688
    qDebug("uid:%" PRIu64 ", groupId:%" PRIu64 " added into table list, slot:%d, total:%d", uid, gid, slot, slot + 1);
10,540✔
2689
  }
2690

2691
  return code;
10,539✔
2692
}
2693

2694
int32_t tableListGetGroupList(const STableListInfo* pTableList, int32_t ordinalGroupIndex, STableKeyInfo** pKeyInfo,
2,797,368✔
2695
                              int32_t* size) {
2696
  int32_t totalGroups = tableListGetOutputGroups(pTableList);
2,797,368✔
2697
  int32_t numOfTables = 0;
2,801,197✔
2698
  int32_t code = tableListGetSize(pTableList, &numOfTables);
2,801,197✔
2699
  if (code != TSDB_CODE_SUCCESS) {
2,800,700!
2700
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
2701
    return code;
×
2702
  }
2703

2704
  if (ordinalGroupIndex < 0 || ordinalGroupIndex >= totalGroups) {
2,800,700!
2705
    return TSDB_CODE_INVALID_PARA;
×
2706
  }
2707

2708
  // here handle two special cases:
2709
  // 1. only one group exists, and 2. one table exists for each group.
2710
  if (totalGroups == 1) {
2,800,700!
2711
    *size = numOfTables;
2,801,612✔
2712
    *pKeyInfo = (*size == 0) ? NULL : taosArrayGet(pTableList->pTableList, 0);
2,801,612✔
2713
    return TSDB_CODE_SUCCESS;
2,798,555✔
2714
  } else if (totalGroups == numOfTables) {
×
2715
    *size = 1;
1,024✔
2716
    *pKeyInfo = taosArrayGet(pTableList->pTableList, ordinalGroupIndex);
1,024✔
2717
    return TSDB_CODE_SUCCESS;
1,024✔
2718
  }
2719

2720
  int32_t offset = pTableList->groupOffset[ordinalGroupIndex];
×
2721
  if (ordinalGroupIndex < totalGroups - 1) {
×
2722
    *size = pTableList->groupOffset[ordinalGroupIndex + 1] - offset;
524✔
2723
  } else {
2724
    *size = numOfTables - offset;
×
2725
  }
2726

2727
  *pKeyInfo = taosArrayGet(pTableList->pTableList, offset);
×
2728
  return TSDB_CODE_SUCCESS;
654✔
2729
}
2730

2731
int32_t tableListGetOutputGroups(const STableListInfo* pTableList) { return pTableList->numOfOuputGroups; }
8,366,532✔
2732

2733
bool oneTableForEachGroup(const STableListInfo* pTableList) { return pTableList->oneTableForEachGroup; }
16,139✔
2734

2735
STableListInfo* tableListCreate() {
4,962,365✔
2736
  STableListInfo* pListInfo = taosMemoryCalloc(1, sizeof(STableListInfo));
4,962,365!
2737
  if (pListInfo == NULL) {
4,976,146!
2738
    return NULL;
×
2739
  }
2740

2741
  pListInfo->remainGroups = NULL;
4,976,146✔
2742
  pListInfo->pTableList = taosArrayInit(4, sizeof(STableKeyInfo));
4,976,146✔
2743
  if (pListInfo->pTableList == NULL) {
4,976,285!
2744
    goto _error;
×
2745
  }
2746

2747
  pListInfo->map = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
4,976,285✔
2748
  if (pListInfo->map == NULL) {
4,975,683!
2749
    goto _error;
×
2750
  }
2751

2752
  pListInfo->numOfOuputGroups = 1;
4,975,683✔
2753
  return pListInfo;
4,975,683✔
2754

2755
_error:
×
2756
  tableListDestroy(pListInfo);
×
2757
  return NULL;
×
2758
}
2759

2760
void tableListDestroy(STableListInfo* pTableListInfo) {
4,978,036✔
2761
  if (pTableListInfo == NULL) {
4,978,036✔
2762
    return;
222✔
2763
  }
2764

2765
  taosArrayDestroy(pTableListInfo->pTableList);
4,977,814✔
2766
  taosMemoryFreeClear(pTableListInfo->groupOffset);
4,978,668!
2767

2768
  taosHashCleanup(pTableListInfo->map);
4,978,668✔
2769
  taosHashCleanup(pTableListInfo->remainGroups);
4,978,515✔
2770
  pTableListInfo->pTableList = NULL;
4,978,605✔
2771
  pTableListInfo->map = NULL;
4,978,605✔
2772
  taosMemoryFree(pTableListInfo);
4,978,605!
2773
}
2774

2775
void tableListClear(STableListInfo* pTableListInfo) {
2,511✔
2776
  if (pTableListInfo == NULL) {
2,511!
2777
    return;
×
2778
  }
2779

2780
  taosArrayClear(pTableListInfo->pTableList);
2,511✔
2781
  taosHashClear(pTableListInfo->map);
2,511✔
2782
  taosHashClear(pTableListInfo->remainGroups);
2,512✔
2783
  taosMemoryFree(pTableListInfo->groupOffset);
2,512!
2784
  pTableListInfo->numOfOuputGroups = 1;
2,511✔
2785
  pTableListInfo->oneTableForEachGroup = false;
2,511✔
2786
}
2787

2788
static int32_t orderbyGroupIdComparFn(const void* p1, const void* p2) {
95,687✔
2789
  STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;
95,687✔
2790
  STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;
95,687✔
2791

2792
  if (pInfo1->groupId == pInfo2->groupId) {
95,687✔
2793
    return 0;
167✔
2794
  } else {
2795
    return pInfo1->groupId < pInfo2->groupId ? -1 : 1;
95,520✔
2796
  }
2797
}
2798

2799
static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
952✔
2800
  taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
952✔
2801
  int32_t size = taosArrayGetSize(pTableListInfo->pTableList);
955✔
2802

2803
  SArray* pList = taosArrayInit(4, sizeof(int32_t));
955✔
2804
  if (!pList) {
955!
2805
    return terrno;
×
2806
  }
2807

2808
  STableKeyInfo* pInfo = taosArrayGet(pTableListInfo->pTableList, 0);
955✔
2809
  if (!pInfo) {
955!
2810
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2811
    return terrno;
×
2812
  }
2813
  uint64_t       gid = pInfo->groupId;
955✔
2814

2815
  int32_t start = 0;
955✔
2816
  void*   tmp = taosArrayPush(pList, &start);
955✔
2817
  if (!tmp) {
955!
2818
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2819
    return terrno;
×
2820
  }
2821

2822
  for (int32_t i = 1; i < size; ++i) {
3,001✔
2823
    pInfo = taosArrayGet(pTableListInfo->pTableList, i);
2,044✔
2824
    if (!pInfo) {
2,044✔
2825
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
1!
2826
      return terrno;
1✔
2827
    }
2828
    if (pInfo->groupId != gid) {
2,043✔
2829
      tmp = taosArrayPush(pList, &i);
1,879✔
2830
      if (!tmp) {
1,879!
2831
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2832
        return terrno;
×
2833
      }
2834
      gid = pInfo->groupId;
1,879✔
2835
    }
2836
  }
2837

2838
  pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);
957✔
2839
  pTableListInfo->groupOffset = taosMemoryMalloc(sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
954!
2840
  if (pTableListInfo->groupOffset == NULL) {
955!
2841
    taosArrayDestroy(pList);
×
2842
    return terrno;
×
2843
  }
2844

2845
  memcpy(pTableListInfo->groupOffset, taosArrayGet(pList, 0), sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
955✔
2846
  taosArrayDestroy(pList);
955✔
2847
  return TSDB_CODE_SUCCESS;
954✔
2848
}
2849

2850
int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle* pHandle, SScanPhysiNode* pScanNode,
4,654,522✔
2851
                                    SNodeList* group, bool groupSort, uint8_t* digest, SStorageAPI* pAPI) {
2852
  int32_t code = TSDB_CODE_SUCCESS;
4,654,522✔
2853

2854
  bool   groupByTbname = groupbyTbname(group);
4,654,522✔
2855
  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
4,655,784✔
2856
  if (!numOfTables) {
4,653,135!
2857
    return code;
×
2858
  }
2859
  if (group == NULL || groupByTbname) {
4,653,135✔
2860
    if (tsCountAlwaysReturnValue && QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode) &&
4,587,513✔
2861
        ((STableScanPhysiNode*)pScanNode)->needCountEmptyTable) {
2,535,696✔
2862
      pTableListInfo->remainGroups =
122,408✔
2863
          taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
122,436✔
2864
      if (pTableListInfo->remainGroups == NULL) {
122,408!
2865
        return terrno;
×
2866
      }
2867

2868
      for (int i = 0; i < numOfTables; i++) {
558,915✔
2869
        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
436,283✔
2870
        if (!info) {
436,269!
2871
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2872
          return terrno;
×
2873
        }
2874
        info->groupId = groupByTbname ? info->uid : 0;
436,282✔
2875

2876
        int32_t tempRes = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId),
436,282✔
2877
                                      &(info->uid), sizeof(info->uid));
436,282✔
2878
        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
436,507!
2879
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
2880
          return tempRes;
×
2881
        }
2882
      }
2883
    } else {
2884
      for (int32_t i = 0; i < numOfTables; i++) {
16,141,731✔
2885
        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
11,676,538✔
2886
        if (!info) {
11,677,275✔
2887
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
621!
2888
          return terrno;
621✔
2889
        }
2890
        info->groupId = groupByTbname ? info->uid : 0;
11,676,654✔
2891
      }
2892
    }
2893

2894
    pTableListInfo->oneTableForEachGroup = groupByTbname;
4,587,825✔
2895
    if (numOfTables == 1 && pTableListInfo->idInfo.tableType == TSDB_CHILD_TABLE) {
4,587,825✔
2896
      pTableListInfo->oneTableForEachGroup = true;
322,286✔
2897
    }
2898

2899
    if (groupSort && groupByTbname) {
4,587,825✔
2900
      taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
29,570✔
2901
      pTableListInfo->numOfOuputGroups = numOfTables;
29,584✔
2902
    } else if (groupByTbname && pScanNode->groupOrderScan) {
4,558,255✔
2903
      pTableListInfo->numOfOuputGroups = numOfTables;
137✔
2904
    } else {
2905
      pTableListInfo->numOfOuputGroups = 1;
4,558,118✔
2906
    }
2907
  } else {
2908
    bool initRemainGroups = false;
65,622✔
2909
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {
65,622✔
2910
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;
64,976✔
2911
      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&
64,976✔
2912
          !(groupSort || pScanNode->groupOrderScan)) {
13,097!
2913
        initRemainGroups = true;
13,053✔
2914
      }
2915
    }
2916

2917
    code = getColInfoResultForGroupby(pHandle->vnode, group, pTableListInfo, digest, pAPI, initRemainGroups);
65,622✔
2918
    if (code != TSDB_CODE_SUCCESS) {
65,705✔
2919
      return code;
14✔
2920
    }
2921

2922
    if (pScanNode->groupOrderScan) pTableListInfo->numOfOuputGroups = taosArrayGetSize(pTableListInfo->pTableList);
65,691✔
2923

2924
    if (groupSort || pScanNode->groupOrderScan) {
65,691✔
2925
      code = sortTableGroup(pTableListInfo);
935✔
2926
    }
2927
  }
2928

2929
  // add all table entry in the hash map
2930
  size_t size = taosArrayGetSize(pTableListInfo->pTableList);
4,653,549✔
2931
  for (int32_t i = 0; i < size; ++i) {
17,054,490✔
2932
    STableKeyInfo* p = taosArrayGet(pTableListInfo->pTableList, i);
12,386,539✔
2933
    if (!p) {
12,371,211!
2934
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2935
      return terrno;
×
2936
    }
2937
    int32_t        tempRes = taosHashPut(pTableListInfo->map, &p->uid, sizeof(uint64_t), &i, sizeof(int32_t));
12,371,211✔
2938
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
12,401,500!
2939
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
2940
      return tempRes;
×
2941
    }
2942
  }
2943

2944
  return code;
4,667,951✔
2945
}
2946

2947
int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
4,873,479✔
2948
                                STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
2949
                                SExecTaskInfo* pTaskInfo) {
2950
  int64_t     st = taosGetTimestampUs();
4,876,991✔
2951
  const char* idStr = GET_TASKID(pTaskInfo);
4,876,991✔
2952

2953
  if (pHandle == NULL) {
4,876,991!
2954
    qError("invalid handle, in creating operator tree, %s", idStr);
×
2955
    return TSDB_CODE_INVALID_PARA;
×
2956
  }
2957

2958
  uint8_t digest[17] = {0};
4,876,991✔
2959
  int32_t code = getTableList(pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, digest, idStr,
4,876,991✔
2960
                              &pTaskInfo->storageAPI);
2961
  if (code != TSDB_CODE_SUCCESS) {
4,879,347✔
2962
    qError("failed to getTableList, code:%s", tstrerror(code));
16!
2963
    return code;
16✔
2964
  }
2965

2966
  int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
4,879,331✔
2967

2968
  int64_t st1 = taosGetTimestampUs();
4,880,755✔
2969
  pTaskInfo->cost.extractListTime = (st1 - st) / 1000.0;
4,880,755✔
2970
  qDebug("extract queried table list completed, %d tables, elapsed time:%.2f ms %s", numOfTables,
4,880,755✔
2971
         pTaskInfo->cost.extractListTime, idStr);
2972

2973
  if (numOfTables == 0) {
4,882,410✔
2974
    qDebug("no table qualified for query, %s" PRIx64, idStr);
221,497✔
2975
    return TSDB_CODE_SUCCESS;
221,501✔
2976
  }
2977

2978
  code = buildGroupIdMapForAllTables(pTableListInfo, pHandle, pScanNode, pGroupTags, groupSort, digest,
4,660,913✔
2979
                                     &pTaskInfo->storageAPI);
2980
  if (code != TSDB_CODE_SUCCESS) {
4,664,115✔
2981
    return code;
14✔
2982
  }
2983

2984
  pTaskInfo->cost.groupIdMapTime = (taosGetTimestampUs() - st1) / 1000.0;
4,663,811✔
2985
  qDebug("generate group id map completed, elapsed time:%.2f ms %s", pTaskInfo->cost.groupIdMapTime, idStr);
4,663,811✔
2986

2987
  return TSDB_CODE_SUCCESS;
4,663,724✔
2988
}
2989

2990
char* getStreamOpName(uint16_t opType) {
1,241,925✔
2991
  switch (opType) {
1,241,925!
2992
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
429,972✔
2993
      return "stream scan";
429,972✔
2994
    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
24,062✔
2995
      return "project";
24,062✔
2996
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:
603,023✔
2997
      return "interval single";
603,023✔
2998
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL:
11,496✔
2999
      return "interval final";
11,496✔
3000
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
35,111✔
3001
      return "interval semi";
35,111✔
3002
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL:
317✔
3003
      return "interval mid";
317✔
3004
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
5,647✔
3005
      return "stream fill";
5,647✔
3006
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:
9,094✔
3007
      return "session single";
9,094✔
3008
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION:
1,162✔
3009
      return "session semi";
1,162✔
3010
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION:
939✔
3011
      return "session final";
939✔
3012
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:
6,821✔
3013
      return "state single";
6,821✔
3014
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
92,010✔
3015
      return "stream partitionby";
92,010✔
3016
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT:
702✔
3017
      return "stream event";
702✔
3018
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT:
3,321✔
3019
      return "stream count";
3,321✔
3020
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC:
9,510✔
3021
      return "stream interp";
9,510✔
3022
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL:
6,339✔
3023
      return "interval continue";
6,339✔
3024
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_INTERVAL:
1,093✔
3025
      return "interval continue semi";
1,093✔
3026
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_INTERVAL:
440✔
3027
      return "interval continue final";
440✔
3028
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SESSION:
×
3029
      return "session continue";
×
3030
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_SESSION:
×
3031
      return "session continue semi";
×
3032
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_SESSION:
×
3033
      return "session continue final";
×
3034
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_STATE:
×
3035
      return "state continue";
×
3036
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_EVENT:
×
3037
      return "event continue";
×
3038
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_COUNT:
×
3039
      return "count continue";
×
3040
  }
3041
  return "error name";
866✔
3042
}
3043

3044
void printDataBlock(SSDataBlock* pBlock, const char* flag, const char* taskIdStr) {
576,261✔
3045
  if (!pBlock) {
576,261✔
3046
    qInfo("%s===stream===%s: Block is Null", taskIdStr, flag);
326!
3047
    return;
326✔
3048
  } else if (pBlock->info.rows == 0) {
575,935✔
3049
    qInfo("%s===stream===%s: Block is Empty. block type %d", taskIdStr, flag, pBlock->info.type);
18,287!
3050
    return;
18,301✔
3051
  }
3052
  if (qDebugFlag & DEBUG_INFO) {
557,648!
3053
    char*   pBuf = NULL;
557,649✔
3054
    int32_t code = dumpBlockData(pBlock, flag, &pBuf, taskIdStr);
557,649✔
3055
    if (code == 0) {
557,656!
3056
      qInfo("%s", pBuf);
557,656!
3057
      taosMemoryFree(pBuf);
557,656!
3058
    }
3059
  }
3060
}
3061

3062
void printSpecDataBlock(SSDataBlock* pBlock, const char* flag, const char* opStr, const char* taskIdStr) {
598,293✔
3063
  if (!pBlock) {
598,293!
3064
    qDebug("%s===stream===%s %s: Block is Null", taskIdStr, flag, opStr);
×
3065
    return;
×
3066
  } else if (pBlock->info.rows == 0) {
598,293✔
3067
    qDebug("%s===stream===%s %s: Block is Empty. block type %d.skey:%" PRId64 ",ekey:%" PRId64 ",version%" PRId64,
14,134✔
3068
           taskIdStr, flag, opStr, pBlock->info.type, pBlock->info.window.skey, pBlock->info.window.ekey,
3069
           pBlock->info.version);
3070
    return;
14,134✔
3071
  }
3072
  if (qDebugFlag & DEBUG_INFO) {
584,159!
3073
    char* pBuf = NULL;
584,165✔
3074
    char  flagBuf[64];
3075
    snprintf(flagBuf, sizeof(flagBuf), "%s %s", flag, opStr);
584,165✔
3076
    int32_t code = dumpBlockData(pBlock, flagBuf, &pBuf, taskIdStr);
584,165✔
3077
    if (code == 0) {
584,157!
3078
      qInfo("%s", pBuf);
584,158!
3079
      taosMemoryFree(pBuf);
584,162!
3080
    }
3081
  }
3082
}
3083

3084
TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }
3,056,274!
3085

3086
void updateTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pWin, int64_t delta) {
420,994,584✔
3087
  int64_t* ts = (int64_t*)pColData->pData;
420,994,584✔
3088

3089
  int64_t duration = pWin->ekey - pWin->skey + delta;
420,994,584✔
3090
  ts[2] = duration;            // set the duration
420,994,584✔
3091
  ts[3] = pWin->skey;          // window start key
420,994,584✔
3092
  ts[4] = pWin->ekey + delta;  // window end key
420,994,584✔
3093
}
420,994,584✔
3094

3095
int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t oldKeysLen, const SSDataBlock* pBlock,
16,489,815✔
3096
                 int32_t rowIndex) {
3097
  SColumnDataAgg* pColAgg = NULL;
16,489,815✔
3098
  const char*     isNull = oldkeyBuf;
16,489,815✔
3099
  const char*     p = oldkeyBuf + sizeof(int8_t) * pSortGroupCols->size;
16,489,815✔
3100

3101
  for (int32_t i = 0; i < pSortGroupCols->size; ++i) {
41,341,270✔
3102
    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
25,330,943✔
3103
    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
25,330,943✔
3104
    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
25,330,943!
3105

3106
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
50,661,886!
3107
      if (isNull[i] != 1) return 1;
1,796,324✔
3108
    } else {
3109
      if (isNull[i] != 0) return 1;
23,534,619✔
3110
      const char* val = colDataGetData(pColInfoData, rowIndex);
23,520,483!
3111
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
23,520,483!
3112
        int32_t len = getJsonValueLen(val);
×
3113
        if (memcmp(p, val, len) != 0) return 1;
×
3114
        p += len;
×
3115
      } else if (IS_VAR_DATA_TYPE(pCol->type)) {
23,520,483!
3116
        if (memcmp(p, val, varDataTLen(val)) != 0) return 1;
7,846,323✔
3117
        p += varDataTLen(val);
7,711,087✔
3118
      } else {
3119
        if (0 != memcmp(p, val, pCol->bytes)) return 1;
15,674,160✔
3120
        p += pCol->bytes;
15,344,204✔
3121
      }
3122
    }
3123
  }
3124
  if ((int32_t)(p - oldkeyBuf) != oldKeysLen) return 1;
16,010,327✔
3125
  return 0;
16,010,101✔
3126
}
3127

3128
int32_t buildKeys(char* keyBuf, const SArray* pSortGroupCols, const SSDataBlock* pBlock, int32_t rowIndex) {
479,512✔
3129
  uint32_t        colNum = pSortGroupCols->size;
479,512✔
3130
  SColumnDataAgg* pColAgg = NULL;
479,512✔
3131
  char*           isNull = keyBuf;
479,512✔
3132
  char*           p = keyBuf + sizeof(int8_t) * colNum;
479,512✔
3133

3134
  for (int32_t i = 0; i < colNum; ++i) {
1,429,866✔
3135
    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
950,354✔
3136
    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
950,354✔
3137
    if (pCol->slotId > pBlock->pDataBlock->size) continue;
950,354!
3138

3139
    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
950,354!
3140

3141
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
1,900,708!
3142
      isNull[i] = 1;
34,376✔
3143
    } else {
3144
      isNull[i] = 0;
915,978✔
3145
      const char* val = colDataGetData(pColInfoData, rowIndex);
915,978!
3146
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
915,978!
3147
        int32_t len = getJsonValueLen(val);
×
3148
        memcpy(p, val, len);
×
3149
        p += len;
×
3150
      } else if (IS_VAR_DATA_TYPE(pCol->type)) {
915,978!
3151
        varDataCopy(p, val);
137,868✔
3152
        p += varDataTLen(val);
137,868✔
3153
      } else {
3154
        memcpy(p, val, pCol->bytes);
778,110✔
3155
        p += pCol->bytes;
778,110✔
3156
      }
3157
    }
3158
  }
3159
  return (int32_t)(p - keyBuf);
479,512✔
3160
}
3161

3162
uint64_t calcGroupId(char* pData, int32_t len) {
17,960,624✔
3163
  T_MD5_CTX context;
3164
  tMD5Init(&context);
17,960,624✔
3165
  tMD5Update(&context, (uint8_t*)pData, len);
17,932,773✔
3166
  tMD5Final(&context);
18,046,438✔
3167

3168
  // NOTE: only extract the initial 8 bytes of the final MD5 digest
3169
  uint64_t id = 0;
18,177,840✔
3170
  memcpy(&id, context.digest, sizeof(uint64_t));
18,177,840✔
3171
  if (0 == id) memcpy(&id, context.digest + 8, sizeof(uint64_t));
18,177,840!
3172
  return id;
18,177,840✔
3173
}
3174

3175
SNodeList* makeColsNodeArrFromSortKeys(SNodeList* pSortKeys) {
736✔
3176
  SNode*     node;
3177
  SNodeList* ret = NULL;
736✔
3178
  FOREACH(node, pSortKeys) {
2,238!
3179
    SOrderByExprNode* pSortKey = (SOrderByExprNode*)node;
1,504✔
3180
    int32_t           code = nodesListMakeAppend(&ret, pSortKey->pExpr);
1,504✔
3181
    if (code != TSDB_CODE_SUCCESS) {
1,502!
3182
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
3183
      terrno = code;
×
3184
      return NULL;
×
3185
    }
3186
  }
3187
  return ret;
734✔
3188
}
3189

3190
int32_t extractKeysLen(const SArray* keys, int32_t* pLen) {
736✔
3191
  int32_t code = TSDB_CODE_SUCCESS;
736✔
3192
  int32_t lino = 0;
736✔
3193
  int32_t len = 0;
736✔
3194
  int32_t keyNum = taosArrayGetSize(keys);
736✔
3195
  for (int32_t i = 0; i < keyNum; ++i) {
1,879✔
3196
    SColumn* pCol = (SColumn*)taosArrayGet(keys, i);
1,143✔
3197
    QUERY_CHECK_NULL(pCol, code, lino, _end, terrno);
1,143!
3198
    len += pCol->bytes;
1,143✔
3199
  }
3200
  len += sizeof(int8_t) * keyNum;  // null flag
736✔
3201
  *pLen = len;
736✔
3202

3203
_end:
736✔
3204
  if (code != TSDB_CODE_SUCCESS) {
736!
3205
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
3206
  }
3207
  return code;
736✔
3208
}
3209

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