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

taosdata / TDengine / #3663

19 Mar 2025 09:21AM UTC coverage: 61.664% (-0.6%) from 62.28%
#3663

push

travis-ci

web-flow
docs: add defination of tmq_config_res_t & fix spell error (#30271)

153169 of 318241 branches covered (48.13%)

Branch coverage included in aggregate %.

239405 of 318390 relevant lines covered (75.19%)

5762846.6 hits per line

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

67.68
/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; }
5,732,853!
60
static int64_t getOffset(const SNode* pLimit) { return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->offset) ? -1 : ((SLimitNode*)pLimit)->offset->datum.i; }
5,733,910✔
61
static void    releaseColInfoData(void* pCol);
62

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

68
void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }
12,279✔
69

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

76
  if (entrySize > 0) {
12,345,527!
77
    memset(pResultRow->pEntryInfo, 0, entrySize);
12,349,518✔
78
  }
79
}
12,345,527✔
80

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

86
size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
1,919,149✔
87
  int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow);
1,919,149✔
88

89
  for (int32_t i = 0; i < numOfOutput; ++i) {
11,839,389✔
90
    rowSize += pCtx[i].resDataInfo.interBufSize;
9,920,240✔
91
  }
92

93
  return rowSize;
1,919,149✔
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,533✔
98
  if (inBuf == NULL || pSup == NULL) {
2,533!
99
    qError("invalid input parameters, inBuf:%p, pSup:%p", inBuf, pSup);
9!
100
    return TSDB_CODE_INVALID_PARA;
9✔
101
  }
102
  SqlFunctionCtx *pCtx = pSup->pCtx;
2,524✔
103
  int32_t        *offset = pSup->rowEntryInfoOffset;
2,524✔
104
  SResultRow     *pResultRow  = NULL;
2,524✔
105
  size_t          processedSize = 0;
2,524✔
106
  int32_t         code = TSDB_CODE_SUCCESS;
2,524✔
107

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

120
  for (int32_t i = 0; i < pSup->numOfExprs; ++i) {
42,469✔
121
    int32_t len = *(int32_t*)inBuf;
39,945✔
122
    inBuf += sizeof(int32_t);
39,945✔
123
    processedSize += sizeof(int32_t);
39,945✔
124
    if (pResultRow->version != FUNCTION_RESULT_INFO_VERSION && pCtx->fpSet.decode) {
39,945!
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);
39,945✔
132
    }
133
    inBuf += len;
39,945✔
134
    processedSize += len;
39,945✔
135
  }
136

137
  if (processedSize < inBufSize) {
2,524✔
138
    // stream stores extra data after result row
139
    size_t leftLen = inBufSize - processedSize;
787✔
140
    TAOS_MEMORY_REALLOC(*outBuf, *outBufSize + leftLen);
787!
141
    if (*outBuf == NULL) {
787!
142
      qError("failed to reallocate memory for output buffer, size:%zu", *outBufSize + leftLen);
×
143
      return terrno;
×
144
    }
145
    (void)memcpy(*outBuf + *outBufSize, inBuf, leftLen);
787✔
146
    inBuf += leftLen;
787✔
147
    processedSize += leftLen;
787✔
148
    *outBufSize += leftLen;
787✔
149
  }
150
  return TSDB_CODE_SUCCESS;
2,524✔
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) {
605,016✔
155
  if (pSup == NULL || inBuf == NULL || outBuf == NULL || outBufSize == NULL) {
605,016!
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;
605,016✔
161
  int32_t        *offset = pSup->rowEntryInfoOffset;
605,016✔
162
  SResultRow     *pResultRow = (SResultRow*)inBuf;
605,016✔
163
  size_t          rowSize = getResultRowSize(pCtx, pSup->numOfExprs);
605,016✔
164

165
  if (rowSize > inBufSize) {
605,017!
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;
605,017✔
172
  if (rowSize < inBufSize) {
605,017✔
173
    *outBufSize += inBufSize - rowSize;
1,126✔
174
  }
175

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

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

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

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

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

220
int32_t resultrowComparAsc(const void* p1, const void* p2) {
838,610,550✔
221
  SResKeyPos* pp1 = *(SResKeyPos**)p1;
838,610,550✔
222
  SResKeyPos* pp2 = *(SResKeyPos**)p2;
838,610,550✔
223

224
  if (pp1->groupId == pp2->groupId) {
838,610,550✔
225
    int64_t pts1 = *(int64_t*)pp1->key;
650,849,749✔
226
    int64_t pts2 = *(int64_t*)pp2->key;
650,849,749✔
227

228
    if (pts1 == pts2) {
650,849,749!
229
      return 0;
×
230
    } else {
231
      return pts1 < pts2 ? -1 : 1;
650,849,749✔
232
    }
233
  } else {
234
    return pp1->groupId < pp2->groupId ? -1 : 1;
187,760,801✔
235
  }
236
}
237

238
static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); }
48,255,611✔
239

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

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

254
  void* pData = NULL;
545,805✔
255
  pGroupResInfo->pRows = taosArrayInit(size, POINTER_BYTES);
545,805✔
256
  QUERY_CHECK_NULL(pGroupResInfo->pRows, code, lino, _end, terrno);
546,042!
257

258
  size_t  keyLen = 0;
546,042✔
259
  int32_t iter = 0;
546,042✔
260
  int64_t bufLen = 0, offset = 0;
546,042✔
261

262
  // todo move away and record this during create window
263
  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
74,628,405✔
264
    /*void* key = */ (void)tSimpleHashGetKey(pData, &keyLen);
265
    bufLen += keyLen + sizeof(SResultRowPosition);
74,082,363✔
266
  }
267

268
  pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);
536,637!
269
  QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
546,091!
270

271
  iter = 0;
546,091✔
272
  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
74,452,234✔
273
    void* key = tSimpleHashGetKey(pData, &keyLen);
73,911,428✔
274

275
    SResKeyPos* p = (SResKeyPos*)(pGroupResInfo->pBuf + offset);
73,911,428✔
276

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

283
    offset += keyLen + sizeof(struct SResultRowPosition);
73,906,143✔
284
  }
285

286
  if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
525,170✔
287
    __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc;
79,877✔
288
    size = POINTER_BYTES;
79,877✔
289
    taosSort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), size, fn);
79,877✔
290
  }
291

292
  pGroupResInfo->index = 0;
546,065✔
293

294
_end:
546,065✔
295
  if (code != TSDB_CODE_SUCCESS) {
546,065!
296
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
297
  }
298
  return code;
545,980✔
299
}
300

301
void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList) {
22,742✔
302
  if (pGroupResInfo->pRows != NULL) {
22,742✔
303
    taosArrayDestroy(pGroupResInfo->pRows);
18,809✔
304
  }
305

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

312
bool hasRemainResults(SGroupResInfo* pGroupResInfo) {
2,365,476✔
313
  if (pGroupResInfo->pRows == NULL) {
2,365,476✔
314
    return false;
1,440✔
315
  }
316

317
  return pGroupResInfo->index < taosArrayGetSize(pGroupResInfo->pRows);
2,364,036✔
318
}
319

320
int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {
1,264,540✔
321
  if (pGroupResInfo->pRows == 0) {
1,264,540✔
322
    return 0;
2,529✔
323
  }
324

325
  return (int32_t)taosArrayGetSize(pGroupResInfo->pRows);
1,262,011✔
326
}
327

328
SArray* createSortInfo(SNodeList* pNodeList) {
295,010✔
329
  size_t numOfCols = 0;
295,010✔
330

331
  if (pNodeList != NULL) {
295,010✔
332
    numOfCols = LIST_LENGTH(pNodeList);
294,647!
333
  } else {
334
    numOfCols = 0;
363✔
335
  }
336

337
  SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));
295,010✔
338
  if (pList == NULL) {
295,019!
339
    return pList;
×
340
  }
341

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

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

365
  return pList;
295,144✔
366
}
367

368
SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {
3,878,392✔
369
  int32_t      numOfCols = LIST_LENGTH(pNode->pSlots);
3,878,392!
370
  SSDataBlock* pBlock = NULL;
3,878,392✔
371
  int32_t      code = createDataBlock(&pBlock);
3,878,392✔
372
  if (code) {
3,881,153!
373
    terrno = code;
×
374
    return NULL;
×
375
  }
376

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

382
  for (int32_t i = 0; i < numOfCols; ++i) {
18,798,542✔
383
    SSlotDescNode*  pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
14,913,723✔
384
    if (!pDescNode) {
14,903,955!
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);
14,903,955✔
393
    idata.info.scale = pDescNode->dataType.scale;
14,931,353✔
394
    idata.info.precision = pDescNode->dataType.precision;
14,931,353✔
395
    idata.info.noData = pDescNode->reserve;
14,931,353✔
396

397
    code = blockDataAppendColInfo(pBlock, &idata);
14,931,353✔
398
    if (code != TSDB_CODE_SUCCESS) {
14,916,767!
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;
3,884,819✔
408
}
409

410
int32_t prepareDataBlockBuf(SSDataBlock* pDataBlock, SColMatchInfo* pMatchInfo) {
1,142,153✔
411
  SDataBlockInfo* pBlockInfo = &pDataBlock->info;
1,142,153✔
412

413
  for (int32_t i = 0; i < taosArrayGetSize(pMatchInfo->pList); ++i) {
4,999,842✔
414
    SColMatchItem* pItem = taosArrayGet(pMatchInfo->pList, i);
3,857,371✔
415
    if (!pItem) {
3,859,130✔
416
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
1,049!
417
      return terrno;
1,049✔
418
    }
419

420
    if (pItem->isPk) {
3,858,081✔
421
      SColumnInfoData* pInfoData = taosArrayGet(pDataBlock->pDataBlock, pItem->dstSlotId);
392✔
422
      if (!pInfoData) {
392!
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;
392✔
427
      pBlockInfo->pks[1].type = pInfoData->info.type;
392✔
428

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

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

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

446
      break;
393✔
447
    }
448
  }
449

450
  return TSDB_CODE_SUCCESS;
1,143,079✔
451
}
452

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

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

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

479
    STagVal tagVal = {0};
1,921✔
480
    tagVal.cid = pSColumnNode->colId;
1,921✔
481
    const char* p = mr->pAPI->extractTagVal(mr->me.ctbEntry.pTags, pSColumnNode->node.resType.type, &tagVal);
1,921✔
482
    if (p == NULL) {
1,921!
483
      res->node.resType.type = TSDB_DATA_TYPE_NULL;
×
484
    } else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {
1,921!
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,921!
492
      res->datum.p = taosMemoryCalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1, 1);
1,898!
493
      if (NULL == res->datum.p) {
1,898!
494
        return DEAL_RES_ERROR;
×
495
      }
496
      memcpy(varDataVal(res->datum.p), tagVal.pData, tagVal.nData);
1,898✔
497
      varDataSetLen(res->datum.p, tagVal.nData);
1,898✔
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,921✔
505
    *pNode = (SNode*)res;
1,921✔
506
  } else if (isTbname) {
4,757!
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,678✔
528
}
529

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

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

540
    return TSDB_CODE_SUCCESS;
×
541
  }
542

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

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

566
    return code;
×
567
  }
568

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

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

576
static EDealRes getColumn(SNode** pNode, void* pContext) {
375,167✔
577
  tagFilterAssist* pData = (tagFilterAssist*)pContext;
375,167✔
578
  SColumnNode*     pSColumnNode = NULL;
375,167✔
579
  if (QUERY_NODE_COLUMN == nodeType((*pNode))) {
375,167✔
580
    pSColumnNode = *(SColumnNode**)pNode;
134,400✔
581
  } else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {
240,767✔
582
    SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);
18,695✔
583
    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {
18,695✔
584
      pData->code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pSColumnNode);
8,102✔
585
      if (NULL == pSColumnNode) {
8,106!
586
        return DEAL_RES_ERROR;
×
587
      }
588
      pSColumnNode->colId = -1;
8,106✔
589
      pSColumnNode->colType = COLUMN_TYPE_TBNAME;
8,106✔
590
      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;
8,106✔
591
      pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;
8,106✔
592
      nodesDestroyNode(*pNode);
8,106✔
593
      *pNode = (SNode*)pSColumnNode;
8,105✔
594
    } else {
595
      return DEAL_RES_CONTINUE;
10,593✔
596
    }
597
  } else {
598
    return DEAL_RES_CONTINUE;
222,072✔
599
  }
600

601
  void* data = taosHashGet(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));
142,505✔
602
  if (!data) {
142,470✔
603
    int32_t tempRes =
604
        taosHashPut(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));
112,179✔
605
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
112,297!
606
      return DEAL_RES_ERROR;
×
607
    }
608
    pSColumnNode->slotId = pData->index++;
112,297✔
609
    SColumnInfo cInfo = {.colId = pSColumnNode->colId,
112,297✔
610
                         .type = pSColumnNode->node.resType.type,
112,297✔
611
                         .bytes = pSColumnNode->node.resType.bytes,
112,297✔
612
                         .pk = pSColumnNode->isPk};
112,297✔
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);
112,297✔
617
    if (!tmp) {
112,339!
618
      return DEAL_RES_ERROR;
×
619
    }
620
  } else {
621
    SColumnNode* col = *(SColumnNode**)data;
30,291✔
622
    pSColumnNode->slotId = col->slotId;
30,291✔
623
  }
624

625
  return DEAL_RES_CONTINUE;
142,630✔
626
}
627

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

634
  pColumnData->info.type = pType->type;
110,227✔
635
  pColumnData->info.bytes = pType->bytes;
110,227✔
636
  pColumnData->info.scale = pType->scale;
110,227✔
637
  pColumnData->info.precision = pType->precision;
110,227✔
638

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

646
  pParam->columnData = pColumnData;
110,205✔
647
  pParam->colAlloced = true;
110,205✔
648
  return TSDB_CODE_SUCCESS;
110,205✔
649
}
650

651
static void releaseColInfoData(void* pCol) {
51,918✔
652
  if (pCol) {
51,918!
653
    SColumnInfoData* col = (SColumnInfoData*)pCol;
51,919✔
654
    colDataDestroy(col);
51,919✔
655
    taosMemoryFree(col);
51,915!
656
  }
657
}
51,922✔
658

659
void freeItem(void* p) {
497,121✔
660
  STUidTagInfo* pInfo = p;
497,121✔
661
  if (pInfo->pTagVal != NULL) {
497,121✔
662
    taosMemoryFree(pInfo->pTagVal);
494,534!
663
  }
664
}
497,013✔
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,
40,033✔
715
                                   SStorageAPI* pAPI, bool initRemainGroups) {
716
  int32_t      code = TSDB_CODE_SUCCESS;
40,033✔
717
  int32_t      lino = 0;
40,033✔
718
  SArray*      pBlockList = NULL;
40,033✔
719
  SSDataBlock* pResBlock = NULL;
40,033✔
720
  void*        keyBuf = NULL;
40,033✔
721
  SArray*      groupData = NULL;
40,033✔
722
  SArray*      pUidTagList = NULL;
40,033✔
723
  SArray*      tableList = NULL;
40,033✔
724

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

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

737
  ctx.index = 0;
40,058✔
738
  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
40,058✔
739
  if (ctx.cInfoList == NULL) {
40,067!
740
    code = terrno;
×
741
    goto end;
×
742
  }
743

744
  SNode* pNode = NULL;
40,067✔
745
  FOREACH(pNode, group) {
91,976!
746
    nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);
51,917✔
747
    if (TSDB_CODE_SUCCESS != ctx.code) {
51,909!
748
      code = ctx.code;
×
749
      goto end;
×
750
    }
751
    REPLACE_NODE(pNode);
51,909✔
752
  }
753

754
  T_MD5_CTX context = {0};
40,059✔
755
  if (tsTagFilterCache) {
40,059!
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));
40,059✔
781
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
40,035!
782

783
  for (int32_t i = 0; i < rows; ++i) {
226,830✔
784
    STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
186,715✔
785
    QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);
186,696!
786
    STUidTagInfo   info = {.uid = pkeyInfo->uid};
186,696✔
787
    void*          tmp = taosArrayPush(pUidTagList, &info);
186,792✔
788
    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
186,792!
789
  }
790

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

796
  int32_t numOfTables = taosArrayGetSize(pUidTagList);
40,071✔
797
  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);
40,070✔
798
  if (pResBlock == NULL) {
40,059!
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);
40,059✔
807
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
40,060!
808

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

812
  groupData = taosArrayInit(2, POINTER_BYTES);
40,060✔
813
  QUERY_CHECK_NULL(groupData, code, lino, end, terrno);
40,070!
814

815
  FOREACH(pNode, group) {
91,977!
816
    SScalarParam output = {0};
51,914✔
817

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

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

837
    if (nodeType(pNode) == QUERY_NODE_COLUMN) {
51,909✔
838
      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;
48,004✔
839
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);
48,004✔
840
      QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);
47,997!
841
      code = colDataAssign(output.columnData, pColInfo, rows, NULL);
47,997✔
842
    } else if (nodeType(pNode) == QUERY_NODE_VALUE) {
3,905!
843
      continue;
×
844
    } else {
845
      code = scalarCalculate(pNode, pBlockList, &output);
3,905✔
846
    }
847

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

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

857
  int32_t keyLen = 0;
40,063✔
858
  SNode*  node;
859
  FOREACH(node, group) {
91,967✔
860
    SExprNode* pExpr = (SExprNode*)node;
51,904✔
861
    keyLen += pExpr->resType.bytes;
51,904✔
862
  }
863

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

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

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

882
  for (int i = 0; i < rows; i++) {
226,810✔
883
    STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
186,732✔
884
    QUERY_CHECK_NULL(info, code, lino, end, terrno);
186,698!
885

886
    char* isNull = (char*)keyBuf;
186,705✔
887
    char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(group);
186,705!
888
    for (int j = 0; j < taosArrayGetSize(groupData); j++) {
402,568✔
889
      SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);
215,890✔
890

891
      if (colDataIsNull_s(pValue, i)) {
431,756✔
892
        isNull[j] = 1;
494✔
893
      } else {
894
        isNull[j] = 0;
215,384✔
895
        char* data = colDataGetData(pValue, i);
215,384!
896
        if (pValue->info.type == TSDB_DATA_TYPE_JSON) {
215,384✔
897
          if (tTagIsJson(data)) {
264✔
898
            code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
12✔
899
            goto end;
12✔
900
          }
901
          if (tTagIsJsonNull(data)) {
252!
902
            isNull[j] = 1;
×
903
            continue;
×
904
          }
905
          int32_t len = getJsonValueLen(data);
252✔
906
          memcpy(pStart, data, len);
252✔
907
          pStart += len;
252✔
908
        } else if (IS_VAR_DATA_TYPE(pValue->info.type)) {
215,120!
909
          if (varDataTLen(data) > pValue->info.bytes) {
184,881✔
910
            code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
3✔
911
            goto end;
3✔
912
          }
913
          memcpy(pStart, data, varDataTLen(data));
184,878✔
914
          pStart += varDataTLen(data);
184,878✔
915
        } else {
916
          memcpy(pStart, data, pValue->info.bytes);
30,239✔
917
          pStart += pValue->info.bytes;
30,239✔
918
        }
919
      }
920
    }
921

922
    int32_t len = (int32_t)(pStart - (char*)keyBuf);
185,669✔
923
    info->groupId = calcGroupId(keyBuf, len);
185,669✔
924
    if (initRemainGroups) {
186,756✔
925
      // groupId ~ table uid
926
      code = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid),
76,440✔
927
                         sizeof(info->uid));
928
      if (code == TSDB_CODE_DUP_KEY) {
76,437✔
929
        code = TSDB_CODE_SUCCESS;
1,229✔
930
      }
931
      QUERY_CHECK_CODE(code, lino, end);
76,437!
932
    }
933
  }
934

935
  if (tsTagFilterCache) {
40,078!
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:
40,078✔
949
  taosMemoryFreeClear(keyBuf);
40,093!
950
  taosHashCleanup(ctx.colHash);
40,088✔
951
  taosArrayDestroy(ctx.cInfoList);
40,058✔
952
  blockDataDestroy(pResBlock);
40,059✔
953
  taosArrayDestroy(pBlockList);
40,068✔
954
  taosArrayDestroyEx(pUidTagList, freeItem);
40,057✔
955
  taosArrayDestroyP(groupData, releaseColInfoData);
40,068✔
956
  if (code != TSDB_CODE_SUCCESS) {
40,065✔
957
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
15!
958
  }
959
  return code;
40,064✔
960
}
961

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

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

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

980
  SArray* pTbList = taosArrayInit(len, POINTER_BYTES);
2,867✔
981
  QUERY_CHECK_NULL(pTbList, code, lino, _end, terrno);
2,868!
982

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

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

997
  size_t numOfTables = taosArrayGetSize(pTbList);
2,867✔
998

999
  // order the name
1000
  taosArraySort(pTbList, nameComparFn);
2,868✔
1001

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

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

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

1022
_end:
2,870✔
1023
  taosArrayDestroy(pTbList);
2,870✔
1024
  if (code != TSDB_CODE_SUCCESS) {
2,868!
1025
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1026
    return NULL;
×
1027
  }
1028
  return pNewList;
2,868✔
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) {
150✔
1043
  STUidTagInfo* p1 = (STUidTagInfo*)a;
150✔
1044
  STUidTagInfo* p2 = (STUidTagInfo*)b;
150✔
1045

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

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

1053
static FilterCondType checkTagCond(SNode* cond) {
58,674✔
1054
  if (nodeType(cond) == QUERY_NODE_OPERATOR) {
58,674✔
1055
    return FILTER_NO_LOGIC;
30,299✔
1056
  }
1057
  if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
28,375!
1058
    return FILTER_AND;
5,083✔
1059
  }
1060
  return FILTER_OTHER;
23,292✔
1061
}
1062

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

1067
  if (ntype == QUERY_NODE_OPERATOR) {
58,737✔
1068
    ret = optimizeTbnameInCondImpl(pVnode, list, cond, pAPI, suid);
30,310✔
1069
  }
1070

1071
  if (ntype != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
58,847✔
1072
    return ret;
35,543✔
1073
  }
1074

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

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

1084
  SListCell* cell = pList->pHead;
23,304✔
1085
  for (int i = 0; i < len; i++) {
70,470✔
1086
    if (cell == NULL) break;
47,189!
1087
    if (optimizeTbnameInCondImpl(pVnode, list, cell->pNode, pAPI, suid) == 0) {
47,189✔
1088
      hasTbnameCond = true;
35✔
1089
      break;
35✔
1090
    }
1091
    cell = cell->pNext;
47,166✔
1092
  }
1093

1094
  taosArraySort(list, filterTableInfoCompare);
23,316✔
1095
  taosArrayRemoveDuplicate(list, filterTableInfoCompare, NULL);
23,184✔
1096

1097
  if (hasTbnameCond) {
23,211✔
1098
    ret = pAPI->metaFn.getTableTagsByUid(pVnode, suid, list);
35✔
1099
  }
1100

1101
  return ret;
23,218✔
1102
}
1103

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

1111
  SOperatorNode* pNode = (SOperatorNode*)pTagCond;
77,351✔
1112
  if (pNode->opType != OP_TYPE_IN) {
77,351✔
1113
    return -1;
70,628✔
1114
  }
1115

1116
  if ((pNode->pLeft != NULL && nodeType(pNode->pLeft) == QUERY_NODE_COLUMN &&
6,723!
1117
       ((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME) &&
6,849✔
1118
      (pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {
2,868!
1119
    SNodeListNode* pList = (SNodeListNode*)pNode->pRight;
2,867✔
1120

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

1126
    SArray*   pTbList = getTableNameList(pList);
2,867✔
1127
    int32_t   numOfTables = taosArrayGetSize(pTbList);
2,867✔
1128
    SHashObj* uHash = NULL;
2,867✔
1129

1130
    size_t numOfExisted = taosArrayGetSize(pExistedUidList);  // len > 0 means there already have uids
2,867✔
1131
    if (numOfExisted > 0) {
2,865!
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++) {
7,427✔
1153
      char* name = taosArrayGetP(pTbList, i);
4,579✔
1154

1155
      uint64_t uid = 0, csuid = 0;
4,578✔
1156
      if (pStoreAPI->metaFn.getTableUidByName(pVnode, name, &uid) == 0) {
4,578✔
1157
        ETableType tbType = TSDB_TABLE_MAX;
2,871✔
1158
        if (pStoreAPI->metaFn.getTableTypeSuidByName(pVnode, name, &tbType, &csuid) == 0 && tbType == TSDB_CHILD_TABLE) {
2,871!
1159
          if (suid != csuid) {
2,850✔
1160
            continue;
20✔
1161
          }
1162
          if (NULL == uHash || taosHashGet(uHash, &uid, sizeof(uid)) == NULL) {
2,830!
1163
            STUidTagInfo s = {.uid = uid, .name = name, .pTagVal = NULL};
2,830✔
1164
            void*        tmp = taosArrayPush(pExistedUidList, &s);
2,831✔
1165
            if (!tmp) {
2,831!
1166
              return terrno;
×
1167
            }
1168
          }
1169
        } else {
1170
          taosArrayDestroy(pTbList);
22✔
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;
1,710✔
1177
      }
1178
    }
1179

1180
    taosHashCleanup(uHash);
2,848✔
1181
    taosArrayDestroy(pTbList);
2,846✔
1182
    return 0;
2,846✔
1183
  }
1184

1185
  return -1;
3,856✔
1186
}
1187

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

1196
  for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {
240,516✔
1197
    SColumnInfoData colInfo = {0};
126,950✔
1198
    void* tmp = taosArrayGet(pColList, i);
126,950✔
1199
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
126,916!
1200
    colInfo.info = *(SColumnInfo*)tmp;
126,916✔
1201
    code = blockDataAppendColInfo(pResBlock, &colInfo);
126,916✔
1202
    QUERY_CHECK_CODE(code, lino, _end);
126,949!
1203
  }
1204

1205
  code = blockDataEnsureCapacity(pResBlock, numOfTables);
113,509✔
1206
  if (code != TSDB_CODE_SUCCESS) {
113,472!
1207
    terrno = code;
×
1208
    blockDataDestroy(pResBlock);
×
1209
    return NULL;
×
1210
  }
1211

1212
  pResBlock->info.rows = numOfTables;
113,472✔
1213

1214
  int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);
113,472✔
1215

1216
  for (int32_t i = 0; i < numOfTables; i++) {
629,634✔
1217
    STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);
515,882✔
1218
    QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
515,606✔
1219

1220
    for (int32_t j = 0; j < numOfCols; j++) {
1,072,892✔
1221
      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, j);
556,782✔
1222
      QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
555,651!
1223

1224
      if (pColInfo->info.colId == -1) {  // tbname
555,867✔
1225
        char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
19,324✔
1226
        if (p1->name != NULL) {
19,324✔
1227
          STR_TO_VARSTR(str, p1->name);
2,831✔
1228
        } else {  // name is not retrieved during filter
1229
          code = pStorageAPI->metaFn.getTableNameByUid(pVnode, p1->uid, str);
16,493✔
1230
          QUERY_CHECK_CODE(code, lino, _end);
16,541!
1231
        }
1232

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

1246
          if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {
538,528✔
1247
            colDataSetNULL(pColInfo, i);
3,068✔
1248
          } else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) {
535,460✔
1249
            code = colDataSetVal(pColInfo, i, p, false);
4,860✔
1250
            QUERY_CHECK_CODE(code, lino, _end);
4,860!
1251
          } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
845,803!
1252
            char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);
315,657✔
1253
            QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
314,923!
1254
            varDataSetLen(tmp, tagVal.nData);
314,923✔
1255
            memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);
314,923✔
1256
            code = colDataSetVal(pColInfo, i, tmp, false);
314,923✔
1257
#if TAG_FILTER_DEBUG
1258
            qDebug("tagfilter varch:%s", tmp + 2);
1259
#endif
1260
            taosMemoryFree(tmp);
315,279!
1261
            QUERY_CHECK_CODE(code, lino, _end);
315,203!
1262
          } else {
1263
            code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false);
214,943✔
1264
            QUERY_CHECK_CODE(code, lino, _end);
214,978!
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:
113,752✔
1279
  if (code != TSDB_CODE_SUCCESS) {
113,752!
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;
113,752✔
1286
}
1287

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

1292
  STableKeyInfo info = {.uid = 0, .groupId = 0};
57,967✔
1293
  int32_t       numOfTables = taosArrayGetSize(pUidTagList);
57,967✔
1294
  for (int32_t i = 0; i < numOfTables; ++i) {
368,021✔
1295
    if (pResultList[i]) {
309,769✔
1296
      STUidTagInfo* tmpTag = (STUidTagInfo*)taosArrayGet(pUidTagList, i);
209,178✔
1297
      if (!tmpTag) {
208,991!
1298
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1299
        return terrno;
×
1300
      }
1301
      uint64_t uid = tmpTag->uid;
208,991✔
1302
      qDebug("tagfilter get uid:%" PRId64 ", res:%d", uid, pResultList[i]);
208,991✔
1303

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

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

1319
  return TSDB_CODE_SUCCESS;
58,252✔
1320
}
1321

1322
static int32_t copyExistedUids(SArray* pUidTagList, const SArray* pUidList) {
58,711✔
1323
  int32_t code = TSDB_CODE_SUCCESS;
58,711✔
1324
  int32_t numOfExisted = taosArrayGetSize(pUidList);
58,711✔
1325
  if (numOfExisted == 0) {
58,775✔
1326
    return code;
55,802✔
1327
  }
1328

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

1345
static int32_t doFilterByTagCond(STableListInfo* pListInfo, SArray* pUidList, SNode* pTagCond, void* pVnode,
1,102,908✔
1346
                                 SIdxFltStatus status, SStorageAPI* pAPI, bool addUid, bool* listAdded) {
1347
  *listAdded = false;
1,102,908✔
1348
  if (pTagCond == NULL) {
1,102,908✔
1349
    return TSDB_CODE_SUCCESS;
1,044,393✔
1350
  }
1351

1352
  terrno = TSDB_CODE_SUCCESS;
58,515✔
1353

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

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

1368
  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
58,896✔
1369
  if (ctx.cInfoList == NULL) {
58,911!
1370
    code = terrno;
×
1371
    QUERY_CHECK_CODE(code, lino, end);
×
1372
  }
1373

1374
  nodesRewriteExprPostOrder(&pTagCond, getColumn, (void*)&ctx);
58,911✔
1375
  if (TSDB_CODE_SUCCESS != ctx.code) {
58,858!
1376
    terrno = code = ctx.code;
×
1377
    goto end;
×
1378
  }
1379

1380
  SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};
58,858✔
1381

1382
  //  int64_t stt = taosGetTimestampUs();
1383
  pUidTagList = taosArrayInit(10, sizeof(STUidTagInfo));
58,858✔
1384
  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
58,754!
1385

1386
  code = copyExistedUids(pUidTagList, pUidList);
58,754✔
1387
  QUERY_CHECK_CODE(code, lino, end);
58,833!
1388

1389
  FilterCondType condType = checkTagCond(pTagCond);
58,833✔
1390

1391
  int32_t filter = optimizeTbnameInCond(pVnode, pListInfo->idInfo.suid, pUidTagList, pTagCond, pAPI);
58,817✔
1392
  if (filter == 0) {  // tbname in filter is activated, do nothing and return
58,674✔
1393
    taosArrayClear(pUidList);
2,847✔
1394

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

1399
    for (int32_t i = 0; i < numOfRows; ++i) {
5,678✔
1400
      STUidTagInfo* pInfo = taosArrayGet(pUidTagList, i);
2,831✔
1401
      QUERY_CHECK_NULL(pInfo, code, lino, end, terrno);
2,831!
1402
      void*         tmp = taosArrayPush(pUidList, &pInfo->uid);
2,831✔
1403
      QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
2,831!
1404
    }
1405
    terrno = 0;
2,847✔
1406
  } else {
1407
    if ((condType == FILTER_NO_LOGIC || condType == FILTER_AND) && status != SFLT_NOT_INDEX) {
55,827✔
1408
      code = pAPI->metaFn.getTableTagsByUid(pVnode, pListInfo->idInfo.suid, pUidTagList);
165✔
1409
    } else {
1410
      code = pAPI->metaFn.getTableTags(pVnode, pListInfo->idInfo.suid, pUidTagList);
55,662✔
1411
    }
1412
    if (code != TSDB_CODE_SUCCESS) {
56,080!
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);
58,927✔
1420
  if (numOfTables == 0) {
58,909✔
1421
    goto end;
606✔
1422
  }
1423

1424
  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);
58,303✔
1425
  if (pResBlock == NULL) {
58,194!
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);
58,194✔
1433
  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
58,295!
1434

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

1438
  code = createResultData(&type, numOfTables, &output);
58,255✔
1439
  if (code != TSDB_CODE_SUCCESS) {
58,296!
1440
    terrno = code;
×
1441
    QUERY_CHECK_CODE(code, lino, end);
×
1442
  }
1443

1444
  code = scalarCalculate(pTagCond, pBlockList, &output);
58,296✔
1445
  if (code != TSDB_CODE_SUCCESS) {
58,111✔
1446
    qError("failed to calculate scalar, reason:%s", tstrerror(code));
8!
1447
    terrno = code;
8✔
1448
    QUERY_CHECK_CODE(code, lino, end);
8!
1449
  }
1450

1451
  code = doSetQualifiedUid(pListInfo, pUidList, pUidTagList, (bool*)output.columnData->pData, addUid);
58,103✔
1452
  if (code != TSDB_CODE_SUCCESS) {
58,245✔
1453
    terrno = code;
65✔
1454
    QUERY_CHECK_CODE(code, lino, end);
×
1455
  }
1456
  *listAdded = true;
58,180✔
1457

1458
end:
58,794✔
1459
  if (code != TSDB_CODE_SUCCESS) {
58,794✔
1460
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
8!
1461
  }
1462
  taosHashCleanup(ctx.colHash);
58,794✔
1463
  taosArrayDestroy(ctx.cInfoList);
58,862✔
1464
  blockDataDestroy(pResBlock);
58,889✔
1465
  taosArrayDestroy(pBlockList);
58,908✔
1466
  taosArrayDestroyEx(pUidTagList, freeItem);
58,910✔
1467

1468
  colDataDestroy(output.columnData);
58,919✔
1469
  taosMemoryFreeClear(output.columnData);
58,921!
1470
  return code;
58,906✔
1471
}
1472

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

1480
  pListInfo->idInfo.suid = pScanNode->suid;
1,100,825✔
1481
  pListInfo->idInfo.tableType = pScanNode->tableType;
1,100,825✔
1482

1483
  SArray* pUidList = taosArrayInit(8, sizeof(uint64_t));
1,100,825✔
1484
  QUERY_CHECK_NULL(pUidList, code, lino, _error, terrno);
1,103,387✔
1485

1486
  SIdxFltStatus status = SFLT_NOT_INDEX;
1,102,561✔
1487
  if (pScanNode->tableType != TSDB_SUPER_TABLE) {
1,102,561✔
1488
    pListInfo->idInfo.uid = pScanNode->uid;
438,153✔
1489
    if (pStorageAPI->metaFn.isTableExisted(pVnode, pScanNode->uid)) {
438,153✔
1490
      void* tmp = taosArrayPush(pUidList, &pScanNode->uid);
436,388✔
1491
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
436,381!
1492
    }
1493
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, false, &listAdded);
438,151✔
1494
    QUERY_CHECK_CODE(code, lino, _end);
438,144!
1495
  } else {
1496
    T_MD5_CTX context = {0};
664,408✔
1497

1498
    if (tsTagFilterCache) {
664,408!
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
664,408✔
1517
      code = pStorageAPI->metaFn.getChildTableList(pVnode, pScanNode->suid, pUidList);
607,626✔
1518
      QUERY_CHECK_CODE(code, lino, _error);
609,400!
1519
    } else {
1520
      // failed to find the result in the cache, let try to calculate the results
1521
      if (pTagIndexCond) {
56,782✔
1522
        void* pIndex = pStorageAPI->metaFn.getInvertIndex(pVnode);
8,071✔
1523

1524
        SIndexMetaArg metaArg = {.metaEx = pVnode,
16,153✔
1525
                                 .idx = pStorageAPI->metaFn.storeGetIndexInfo(pVnode),
8,075✔
1526
                                 .ivtIdx = pIndex,
1527
                                 .suid = pScanNode->uid};
8,078✔
1528

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

1539
    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, tsTagFilterCache, &listAdded);
666,174✔
1540
    QUERY_CHECK_CODE(code, lino, _end);
664,920✔
1541

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

1545
    if (tsTagFilterCache) {
664,874✔
1546
      size_t size = numOfTables * sizeof(uint64_t) + sizeof(int32_t);
69✔
1547
      char*  pPayload = taosMemoryMalloc(size);
69!
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:
1,102,957✔
1567
  if (!listAdded) {
1,102,957✔
1568
    numOfTables = taosArrayGetSize(pUidList);
1,044,900✔
1569
    for (int i = 0; i < numOfTables; i++) {
3,923,337✔
1570
      void* tmp = taosArrayGet(pUidList, i);
2,877,736✔
1571
      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
2,875,845!
1572
      STableKeyInfo info = {.uid = *(uint64_t*)tmp, .groupId = 0};
2,875,845✔
1573

1574
      void* p = taosArrayPush(pListInfo->pTableList, &info);
2,875,845✔
1575
      if (p == NULL) {
2,878,164!
1576
        taosArrayDestroy(pUidList);
×
1577
        return terrno;
×
1578
      }
1579

1580
      qTrace("tagfilter get uid:%" PRIu64 ", %s", info.uid, idstr);
2,878,164✔
1581
    }
1582
  }
1583

1584
_error:
1,103,658✔
1585
  taosArrayDestroy(pUidList);
1,103,658✔
1586
  if (code != TSDB_CODE_SUCCESS) {
1,103,538✔
1587
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
8!
1588
  }
1589
  return code;
1,103,008✔
1590
}
1591

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

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

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

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

1622
  SNode* node;
1623
  FOREACH(node, pGroups) {
54,125!
1624
    SExprNode* pExpr = (SExprNode*)node;
44,840✔
1625
    keyLen += pExpr->resType.bytes;
44,840✔
1626
  }
1627

1628
  keyLen += sizeof(int8_t) * LIST_LENGTH(pGroups);
9,285!
1629
  return keyLen;
9,285✔
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) {
25,272✔
1716
  if (!pNodeList) {
25,272!
1717
    return NULL;
×
1718
  }
1719

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

1726
  for (int32_t i = 0; i < numOfCols; ++i) {
102,009✔
1727
    SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i);
76,713✔
1728
    if (!pColNode) {
76,679!
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};
76,679✔
1736
    c.slotId = pColNode->slotId;
76,679✔
1737
    c.colId = pColNode->colId;
76,679✔
1738
    c.type = pColNode->node.resType.type;
76,679✔
1739
    c.bytes = pColNode->node.resType.bytes;
76,679✔
1740
    c.precision = pColNode->node.resType.precision;
76,679✔
1741
    c.scale = pColNode->node.resType.scale;
76,679✔
1742

1743
    void* tmp = taosArrayPush(pList, &c);
76,743✔
1744
    if (!tmp) {
76,743!
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;
25,296✔
1752
}
1753

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

1760
  pMatchInfo->matchType = type;
1,554,538✔
1761

1762
  SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchItem));
1,554,538✔
1763
  if (pList == NULL) {
1,555,312!
1764
    code = terrno;
×
1765
    return code;
×
1766
  }
1767

1768
  for (int32_t i = 0; i < numOfCols; ++i) {
7,357,253✔
1769
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);
5,799,666✔
1770
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
5,788,177!
1771
    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {
5,788,803✔
1772
      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
5,700,050✔
1773

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

1785
  // set the output flag for each column in SColMatchInfo, according to the
1786
  *numOfOutputCols = 0;
1,557,587✔
1787
  int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);
1,557,587✔
1788
  for (int32_t i = 0; i < num; ++i) {
8,389,197✔
1789
    SSlotDescNode* pNode = (SSlotDescNode*)nodesListGetNode(pOutputNodeList->pSlots, i);
6,828,976✔
1790
    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
6,828,591!
1791

1792
    // todo: add reserve flag check
1793
    // it is a column reserved for the arithmetic expression calculation
1794
    if (pNode->slotId >= numOfCols) {
6,833,006✔
1795
      (*numOfOutputCols) += 1;
1,040,969✔
1796
      continue;
1,040,969✔
1797
    }
1798

1799
    SColMatchItem* info = NULL;
5,792,037✔
1800
    for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {
684,545,127✔
1801
      info = taosArrayGet(pList, j);
686,024,914✔
1802
      QUERY_CHECK_NULL(info, code, lino, _end, terrno);
687,536,681✔
1803
      if (info->dstSlotId == pNode->slotId) {
684,451,527✔
1804
        break;
5,698,437✔
1805
      }
1806
    }
1807

1808
    if (pNode->output) {
5,790,641✔
1809
      (*numOfOutputCols) += 1;
5,719,708✔
1810
    } else if (info != NULL) {
70,933!
1811
      // select distinct tbname from stb where tbname='abc';
1812
      info->needOutput = false;
70,936✔
1813
    }
1814
  }
1815

1816
  pMatchInfo->pList = pList;
1,560,221✔
1817

1818
_end:
1,560,221✔
1819
  if (code != TSDB_CODE_SUCCESS) {
1,560,221!
1820
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1821
  }
1822
  return code;
1,555,397✔
1823
}
1824

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

1835
  return s;
5,743,502✔
1836
}
1837

1838
static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType, EColumnType colType) {
4,979,505✔
1839
  SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn));
4,979,505!
1840
  if (pCol == NULL) {
4,981,153!
1841
    return NULL;
×
1842
  }
1843

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

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

1863
  pExp->pExpr->_function.num = 1;
5,745,712✔
1864
  pExp->pExpr->_function.functionId = -1;
5,745,712✔
1865

1866
  int32_t type = nodeType(pNode);
5,745,712✔
1867
  // it is a project query, or group by column
1868
  if (type == QUERY_NODE_COLUMN) {
5,745,712✔
1869
    pExp->pExpr->nodeType = QUERY_NODE_COLUMN;
2,840,498✔
1870
    SColumnNode* pColNode = (SColumnNode*)pNode;
2,840,498✔
1871

1872
    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
2,840,498!
1873
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
2,840,478!
1874

1875
    pExp->base.numOfParams = 1;
2,840,478✔
1876

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

1881
    pExp->base.pParam[0].pCol =
5,679,649✔
1882
        createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType, pColNode->colType);
2,839,733✔
1883
    QUERY_CHECK_NULL(pExp->base.pParam[0].pCol, code, lino, _end, terrno);
2,839,916!
1884

1885
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN;
2,839,916✔
1886
  } else if (type == QUERY_NODE_VALUE) {
2,905,214✔
1887
    pExp->pExpr->nodeType = QUERY_NODE_VALUE;
169,332✔
1888
    SValueNode* pValNode = (SValueNode*)pNode;
169,332✔
1889

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

1893
    pExp->base.numOfParams = 1;
169,416✔
1894

1895
    SDataType* pType = &pValNode->node.resType;
169,416✔
1896
    pExp->base.resSchema =
1897
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pValNode->node.aliasName);
169,416✔
1898
    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;
169,391✔
1899
    code = nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param);
169,391✔
1900
    QUERY_CHECK_CODE(code, lino, _end);
166,382!
1901
  } else if (type == QUERY_NODE_FUNCTION) {
2,735,882✔
1902
    pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;
2,583,303✔
1903
    SFunctionNode* pFuncNode = (SFunctionNode*)pNode;
2,583,303✔
1904

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

1910
    pExprNode->_function.functionId = pFuncNode->funcId;
2,583,109✔
1911
    pExprNode->_function.pFunctNode = pFuncNode;
2,583,109✔
1912
    pExprNode->_function.functionType = pFuncNode->funcType;
2,583,109✔
1913

1914
    tstrncpy(pExprNode->_function.functionName, pFuncNode->functionName, tListLen(pExprNode->_function.functionName));
2,583,109✔
1915

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

1921
    if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) &&
2,583,109✔
1922
        pExprNode->_function.functionName[len] == 0) {
189,676!
1923
      pFuncNode->pParameterList = NULL;
189,733✔
1924
      int32_t     code = nodesMakeList(&pFuncNode->pParameterList);
189,733✔
1925
      SValueNode* res = NULL;
189,748✔
1926
      if (TSDB_CODE_SUCCESS == code) {
189,748!
1927
        code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
189,755✔
1928
      }
1929
      QUERY_CHECK_CODE(code, lino, _end);
189,653!
1930
      res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
189,653✔
1931
      code = nodesListAppend(pFuncNode->pParameterList, (SNode*)res);
189,653✔
1932
      if (code != TSDB_CODE_SUCCESS) {
189,639!
1933
        nodesDestroyNode((SNode*)res);
×
1934
        res = NULL;
×
1935
      }
1936
      QUERY_CHECK_CODE(code, lino, _end);
189,644!
1937
    }
1938
#endif
1939

1940
    int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);
2,583,020✔
1941

1942
    pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam));
2,583,020!
1943
    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
2,583,779!
1944
    pExp->base.numOfParams = numOfParam;
2,583,779✔
1945

1946
    for (int32_t j = 0; j < numOfParam && TSDB_CODE_SUCCESS == code; ++j) {
5,831,163!
1947
      SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j);
3,247,595✔
1948
      QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
3,247,110!
1949
      if (p1->type == QUERY_NODE_COLUMN) {
3,247,569✔
1950
        SColumnNode* pcn = (SColumnNode*)p1;
2,141,244✔
1951

1952
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN;
2,141,244✔
1953
        pExp->base.pParam[j].pCol =
4,283,069✔
1954
            createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType, pcn->colType);
2,141,244✔
1955
        QUERY_CHECK_NULL(pExp->base.pParam[j].pCol, code, lino, _end, terrno);
2,141,825✔
1956
      } else if (p1->type == QUERY_NODE_VALUE) {
1,106,325✔
1957
        SValueNode* pvn = (SValueNode*)p1;
696,379✔
1958
        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE;
696,379✔
1959
        code = nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param);
696,379✔
1960
        QUERY_CHECK_CODE(code, lino, _end);
696,151!
1961
      }
1962
    }
1963
    pExp->pExpr->_function.bindExprID = ((SExprNode*)pNode)->bindExprID;
2,583,568✔
1964
  } else if (type == QUERY_NODE_OPERATOR) {
152,579✔
1965
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
146,447✔
1966
    SOperatorNode* pOpNode = (SOperatorNode*)pNode;
146,447✔
1967

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

1972
    SDataType* pType = &pOpNode->node.resType;
146,484✔
1973
    pExp->base.resSchema =
1974
        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pOpNode->node.aliasName);
146,484✔
1975
    pExp->pExpr->_optrRoot.pRootNode = pNode;
146,406✔
1976
  } else if (type == QUERY_NODE_CASE_WHEN) {
6,132✔
1977
    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
5,977✔
1978
    SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;
5,977✔
1979

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

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

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

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

2020
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
22,817✔
2021
    SExprInfo* pExp = &pExprs[i];
20,111✔
2022
    int32_t    code = createExprFromOneNode(pExp, nodesListGetNode(pNodeList, i), i + UD_TAG_COLUMN_INDEX);
20,111✔
2023
    if (code != TSDB_CODE_SUCCESS) {
20,112!
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,706✔
2032
}
2033

2034
int32_t createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, SExprInfo** pExprInfo, int32_t* numOfExprs) {
2,235,848✔
2035
  QRY_PARAM_CHECK(pExprInfo);
2,235,848!
2036

2037
  int32_t code = 0;
2,235,848✔
2038
  int32_t numOfFuncs = LIST_LENGTH(pNodeList);
2,235,848✔
2039
  int32_t numOfGroupKeys = 0;
2,235,848✔
2040
  if (pGroupKeys != NULL) {
2,235,848✔
2041
    numOfGroupKeys = LIST_LENGTH(pGroupKeys);
207,639!
2042
  }
2043

2044
  *numOfExprs = numOfFuncs + numOfGroupKeys;
2,235,848✔
2045
  if (*numOfExprs == 0) {
2,235,848✔
2046
    return code;
215,058✔
2047
  }
2048

2049
  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
2,020,790!
2050
  if (pExprs == NULL) {
2,021,875!
2051
    return terrno;
×
2052
  }
2053

2054
  for (int32_t i = 0; i < (*numOfExprs); ++i) {
7,738,046✔
2055
    STargetNode* pTargetNode = NULL;
5,715,989✔
2056
    if (i < numOfFuncs) {
5,715,989✔
2057
      pTargetNode = (STargetNode*)nodesListGetNode(pNodeList, i);
5,466,087✔
2058
    } else {
2059
      pTargetNode = (STargetNode*)nodesListGetNode(pGroupKeys, i - numOfFuncs);
249,902✔
2060
    }
2061
    if (!pTargetNode) {
5,711,078!
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];
5,711,078✔
2069
    code = createExprFromTargetNode(pExp, pTargetNode);
5,711,078✔
2070
    if (code != TSDB_CODE_SUCCESS) {
5,716,935✔
2071
      destroyExprInfo(pExprs, *numOfExprs);
764✔
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;
2,022,057✔
2079
  return code;
2,022,057✔
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) {
2,201,406✔
2091
  int32_t num = 0;
2,201,406✔
2092
  int32_t code = TSDB_CODE_SUCCESS;
2,201,406✔
2093
  int32_t lino = 0;
2,201,406✔
2094

2095
  SArray* pValCtxArray = NULL;
2,201,406✔
2096
  for (int32_t i = numOfOutput - 1; i > 0; --i) {  // select Func is at the end of the list
5,735,831✔
2097
    int32_t funcIdx = pCtx[i].pExpr->pExpr->_function.bindExprID;
3,534,425✔
2098
    if (funcIdx > 0) {
3,534,425✔
2099
      if (pValCtxArray == NULL) {
3,482✔
2100
        // the end of the list is the select function of biggest index
2101
        pValCtxArray = taosArrayInit_s(sizeof(SSubsidiaryResInfo*), funcIdx);
2,472✔
2102
        if (pValCtxArray == NULL) {
2,475✔
2103
          return terrno;
10✔
2104
        }
2105
      }
2106
      if (funcIdx > pValCtxArray->size) {
3,475!
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,475✔
2112
      pSubsidiary->pCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
3,475✔
2113
      if (pSubsidiary->pCtx == NULL) {
3,492!
2114
        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
×
2115
        return terrno;
×
2116
      }
2117
      pSubsidiary->num = 0;
3,492✔
2118
      taosArraySet(pValCtxArray, funcIdx - 1, &pSubsidiary);
3,492✔
2119
    }
2120
  }
2121

2122
  SqlFunctionCtx*  p = NULL;
2,201,406✔
2123
  SqlFunctionCtx** pValCtx = NULL;
2,201,406✔
2124
  if (pValCtxArray == NULL) {
2,201,406✔
2125
    pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
2,198,879!
2126
    if (pValCtx == NULL) {
2,199,998!
2127
      QUERY_CHECK_CODE(terrno, lino, _end);
×
2128
    }
2129
  }
2130

2131
  for (int32_t i = 0; i < numOfOutput; ++i) {
7,728,759✔
2132
    const char* pName = pCtx[i].pExpr->pExpr->_function.functionName;
5,526,501✔
2133
    if ((strcmp(pName, "_select_value") == 0)) {
5,526,501✔
2134
      if (pValCtxArray == NULL) {
31,319✔
2135
        pValCtx[num++] = &pCtx[i];
26,522✔
2136
      } else {
2137
        int32_t bindFuncIndex = pCtx[i].pExpr->pExpr->relatedTo;  // start from index 1;
4,797✔
2138
        if (bindFuncIndex > 0) {  // 0 is default index related to the select function
4,797✔
2139
          bindFuncIndex -= 1;
4,678✔
2140
        }
2141
        SSubsidiaryResInfo** pSubsidiary = taosArrayGet(pValCtxArray, bindFuncIndex);
4,797✔
2142
        if(pSubsidiary == NULL) {
4,793!
2143
          QUERY_CHECK_CODE(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, lino, _end);
×
2144
        }
2145
        (*pSubsidiary)->pCtx[(*pSubsidiary)->num] = &pCtx[i];
4,793✔
2146
        (*pSubsidiary)->num++;
4,793✔
2147
      }
2148
    } else if (fmIsSelectFunc(pCtx[i].functionId)) {
5,495,182✔
2149
       if (pValCtxArray == NULL) {
390,475✔
2150
        p = &pCtx[i];
386,243✔
2151
       }
2152
    }
2153
  }
2154

2155
  if (p != NULL) {
2,202,258✔
2156
    p->subsidiaries.pCtx = pValCtx;
131,363✔
2157
    p->subsidiaries.num = num;
131,363✔
2158
  } else {
2159
    taosMemoryFreeClear(pValCtx);
2,070,895!
2160
  }
2161

2162
_end:
3,615✔
2163
  if (code != TSDB_CODE_SUCCESS) {
2,202,487!
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);
2,202,487✔
2169
  }
2170
  return code;
2,199,176✔
2171
}
2172

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

2182
  *rowEntryInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t));
2,200,527!
2183
  if (*rowEntryInfoOffset == 0) {
2,202,657!
2184
    taosMemoryFreeClear(pFuncCtx);
×
2185
    return NULL;
×
2186
  }
2187

2188
  for (int32_t i = 0; i < numOfOutput; ++i) {
7,737,555✔
2189
    SExprInfo* pExpr = &pExprInfo[i];
5,533,318✔
2190

2191
    SExprBasicInfo* pFunct = &pExpr->base;
5,533,318✔
2192
    SqlFunctionCtx* pCtx = &pFuncCtx[i];
5,533,318✔
2193

2194
    pCtx->functionId = -1;
5,533,318✔
2195
    pCtx->pExpr = pExpr;
5,533,318✔
2196

2197
    if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {
5,533,318✔
2198
      SFuncExecEnv env = {0};
2,579,734✔
2199
      pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;
2,579,734✔
2200
      pCtx->isPseudoFunc = fmIsWindowPseudoColumnFunc(pCtx->functionId);
2,579,734✔
2201
      pCtx->isNotNullFunc = fmIsNotNullOutputFunc(pCtx->functionId);
2,578,188✔
2202

2203
      bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);
2,578,830✔
2204
      if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId)) {
4,185,444✔
2205
        if (!isUdaf) {
1,609,107✔
2206
          code = fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
1,609,075✔
2207
          QUERY_CHECK_CODE(code, lino, _end);
1,608,271!
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);
1,608,303✔
2217
        if (!tmp) {
1,607,069!
2218
          code = terrno;
×
2219
          QUERY_CHECK_CODE(code, lino, _end);
×
2220
        }
2221
      } else {
2222
        code = fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);
968,205✔
2223
        if (code != TSDB_CODE_SUCCESS && isUdaf) {
968,134!
2224
          code = TSDB_CODE_SUCCESS;
262✔
2225
        }
2226
        QUERY_CHECK_CODE(code, lino, _end);
968,134!
2227

2228
        if (pCtx->sfp.getEnv != NULL) {
968,134✔
2229
          bool tmp = pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
200,789✔
2230
          if (!tmp) {
200,766✔
2231
            code = terrno;
584✔
2232
            QUERY_CHECK_CODE(code, lino, _end);
×
2233
          }
2234
        }
2235
      }
2236
      pCtx->resDataInfo.interBufSize = env.calcMemSize;
2,575,204✔
2237
    } else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR ||
2,953,584✔
2238
               pExpr->pExpr->nodeType == QUERY_NODE_VALUE) {
169,360✔
2239
      // for simple column, the result buffer needs to hold at least one element.
2240
      pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes;
2,953,553✔
2241
    }
2242

2243
    pCtx->input.numOfInputCols = pFunct->numOfParams;
5,528,788✔
2244
    pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
5,528,788✔
2245
    QUERY_CHECK_NULL(pCtx->input.pData, code, lino, _end, terrno);
5,534,760!
2246
    pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
5,534,760!
2247
    QUERY_CHECK_NULL(pCtx->input.pColumnDataAgg, code, lino, _end, terrno);
5,534,898!
2248

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

2263
  for (int32_t i = 1; i < numOfOutput; ++i) {
5,739,523✔
2264
    (*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) +
3,535,286✔
2265
                                         pFuncCtx[i - 1].resDataInfo.interBufSize);
3,535,286✔
2266
  }
2267

2268
  code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);
2,204,237✔
2269
  QUERY_CHECK_CODE(code, lino, _end);
2,199,723!
2270

2271
_end:
2,199,723✔
2272
  if (code != TSDB_CODE_SUCCESS) {
2,199,723!
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;
2,199,723✔
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) {
46,612✔
2290
  int32_t code = TSDB_CODE_SUCCESS;
46,612✔
2291
  size_t  numOfSrcCols = taosArrayGetSize(pCols);
46,612✔
2292

2293
  int32_t i = 0, j = 0;
46,609✔
2294
  while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) {
374,015✔
2295
    SColumnInfoData* p = taosArrayGet(pCols, i);
327,419✔
2296
    if (!p) {
327,396!
2297
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2298
      return terrno;
×
2299
    }
2300
    SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);
327,396✔
2301
    if (!pmInfo) {
327,386!
2302
      return terrno;
×
2303
    }
2304

2305
    if (p->info.colId == pmInfo->colId) {
327,424✔
2306
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->dstSlotId);
213,521✔
2307
      if (!pDst) {
213,515!
2308
        return terrno;
×
2309
      }
2310
      code = colDataAssign(pDst, p, pBlock->info.rows, &pBlock->info);
213,515✔
2311
      if (code != TSDB_CODE_SUCCESS) {
213,505✔
2312
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
2!
2313
        return code;
×
2314
      }
2315
      i++;
213,503✔
2316
      j++;
213,503✔
2317
    } else if (p->info.colId < pmInfo->colId) {
113,903!
2318
      i++;
113,903✔
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;
46,615✔
2325
}
2326

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

2339
  return interval;
966,641✔
2340
}
2341

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

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

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

2359
  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
1,141,501!
2360
  if (!pCond->colList) {
1,143,780!
2361
    return terrno;
×
2362
  }
2363
  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
1,143,780!
2364
  if (pCond->pSlotList == NULL) {
1,143,613!
2365
    taosMemoryFreeClear(pCond->colList);
×
2366
    return terrno;
×
2367
  }
2368

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

2377
  // allowed read stt file optimization mode
2378
  pCond->notLoadData = (pTableScanNode->dataRequired == FUNC_DATA_REQUIRED_NOT_LOAD) &&
2,350,581✔
2379
                       (pTableScanNode->scan.node.pConditions == NULL) && (pTableScanNode->interval == 0);
1,143,613✔
2380

2381
  int32_t j = 0;
1,143,613✔
2382
  for (int32_t i = 0; i < pCond->numOfCols; ++i) {
4,997,327✔
2383
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pTableScanNode->scan.pScanCols, i);
3,853,211✔
2384
    if (!pNode) {
3,853,462!
2385
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2386
      return terrno;
×
2387
    }
2388
    SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
3,853,714✔
2389
    if (pColNode->colType == COLUMN_TYPE_TAG) {
3,853,714!
2390
      continue;
×
2391
    }
2392

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

2398
    pCond->pSlotList[j] = pNode->slotId;
3,853,714✔
2399
    j += 1;
3,853,714✔
2400
  }
2401

2402
  pCond->numOfCols = j;
1,144,116✔
2403
  return TSDB_CODE_SUCCESS;
1,144,116✔
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) {
2,468,849✔
2455
  taosMemoryFreeClear(pCond->colList);
2,468,849!
2456
  taosMemoryFreeClear(pCond->pSlotList);
2,469,089!
2457
}
2,469,073✔
2458

2459
int32_t convertFillType(int32_t mode) {
24,754✔
2460
  int32_t type = TSDB_FILL_NONE;
24,754✔
2461
  switch (mode) {
24,754!
2462
    case FILL_MODE_PREV:
844✔
2463
      type = TSDB_FILL_PREV;
844✔
2464
      break;
844✔
2465
    case FILL_MODE_NONE:
×
2466
      type = TSDB_FILL_NONE;
×
2467
      break;
×
2468
    case FILL_MODE_NULL:
10,999✔
2469
      type = TSDB_FILL_NULL;
10,999✔
2470
      break;
10,999✔
2471
    case FILL_MODE_NULL_F:
10✔
2472
      type = TSDB_FILL_NULL_F;
10✔
2473
      break;
10✔
2474
    case FILL_MODE_NEXT:
516✔
2475
      type = TSDB_FILL_NEXT;
516✔
2476
      break;
516✔
2477
    case FILL_MODE_VALUE:
1,055✔
2478
      type = TSDB_FILL_SET_VALUE;
1,055✔
2479
      break;
1,055✔
2480
    case FILL_MODE_VALUE_F:
19✔
2481
      type = TSDB_FILL_SET_VALUE_F;
19✔
2482
      break;
19✔
2483
    case FILL_MODE_LINEAR:
1,283✔
2484
      type = TSDB_FILL_LINEAR;
1,283✔
2485
      break;
1,283✔
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;
24,754✔
2494
}
2495

2496
void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery) {
794,743✔
2497
  if (ascQuery) {
794,743✔
2498
    *w = getAlignQueryTimeWindow(pInterval, ts);
790,935✔
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);
3,808✔
2502

2503
    int64_t key = w->skey;
3,818✔
2504
    while (key < ts) {  // moving towards end
3,884✔
2505
      key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);
2,284✔
2506
      if (key > ts) {
2,286✔
2507
        break;
2,220✔
2508
      }
2509

2510
      w->skey = key;
66✔
2511
    }
2512
    w->ekey = taosTimeAdd(w->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
3,820✔
2513
  }
2514
}
794,763✔
2515

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

2519
  w.skey = taosTimeTruncate(ts, pInterval);
660,277✔
2520
  w.ekey = taosTimeGetIntervalEnd(w.skey, pInterval);
660,423✔
2521
  return w;
660,361✔
2522
}
2523

2524
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {
9,692✔
2525
  STimeWindow win = *pWindow;
9,692✔
2526
  STimeWindow save = win;
9,692✔
2527
  while (win.skey <= ts && win.ekey >= ts) {
79,595✔
2528
    save = win;
69,902✔
2529
    // get previous time window
2530
    getNextTimeWindow(pInterval, &win, order == TSDB_ORDER_ASC ? TSDB_ORDER_DESC : TSDB_ORDER_ASC);
69,902✔
2531
  }
2532

2533
  return save;
9,693✔
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,
1,712,091✔
2539
                                int32_t order) {
2540
  STimeWindow w = {0};
1,712,091✔
2541
  if (pResultRowInfo->cur.pageId == -1) {  // the first window, from the previous stored value
1,712,091✔
2542
    getInitialStartTimeWindow(pInterval, ts, &w, (order == TSDB_ORDER_ASC));
794,746✔
2543
    return w;
794,748✔
2544
  }
2545

2546
  SResultRow* pRow = getResultRowByPos(pBuf, &pResultRowInfo->cur, false);
917,345✔
2547
  if (pRow) {
917,319!
2548
    TAOS_SET_OBJ_ALIGNED(&w, pRow->win);
917,324✔
2549
  }
2550

2551
  // in case of typical time window, we can calculate time window directly.
2552
  if (w.skey > ts || w.ekey < ts) {
917,319✔
2553
    w = doCalculateTimeWindow(ts, pInterval);
660,291✔
2554
  }
2555

2556
  if (pInterval->interval != pInterval->sliding) {
917,386✔
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);
9,692✔
2560
  }
2561

2562
  return w;
917,391✔
2563
}
2564

2565
TSKEY getNextTimeWindowStart(const SInterval* pInterval, TSKEY start, int32_t order) {
104,574,990✔
2566
  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
104,574,990✔
2567
  TSKEY   nextStart = taosTimeAdd(start, -1 * pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
104,574,990✔
2568
  nextStart = taosTimeAdd(nextStart, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision, NULL);
104,550,208✔
2569
  nextStart = taosTimeAdd(nextStart, pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);
104,608,131✔
2570
  return nextStart;
104,281,706✔
2571
}
2572

2573
void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t order) {
104,605,682✔
2574
  tw->skey = getNextTimeWindowStart(pInterval, tw->skey, order);
104,605,682✔
2575
  tw->ekey = taosTimeAdd(tw->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
104,306,940✔
2576
}
104,258,164✔
2577

2578
bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {
3,368,843✔
2579
  return (pLimitInfo->limit.limit != -1 || pLimitInfo->limit.offset != -1 || pLimitInfo->slimit.limit != -1 ||
6,713,609!
2580
          pLimitInfo->slimit.offset != -1);
3,344,766✔
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) {
2,866,531✔
2588
  SLimit limit = {.limit = getLimit(pLimit), .offset = getOffset(pLimit)};
2,866,531✔
2589
  SLimit slimit = {.limit = getLimit(pSLimit), .offset = getOffset(pSLimit)};
2,867,034✔
2590

2591
  pLimitInfo->limit = limit;
2,867,146✔
2592
  pLimitInfo->slimit = slimit;
2,867,146✔
2593
  pLimitInfo->remainOffset = limit.offset;
2,867,146✔
2594
  pLimitInfo->remainGroupOffset = slimit.offset;
2,867,146✔
2595
}
2,867,146✔
2596

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

2602
int32_t tableListGetSize(const STableListInfo* pTableList, int32_t* pRes) {
3,089,467✔
2603
  if (taosArrayGetSize(pTableList->pTableList) != taosHashGetSize(pTableList->map)) {
3,089,467!
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);
3,092,342✔
2608
  return TSDB_CODE_SUCCESS;
3,092,661✔
2609
}
2610

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

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

2618
  return taosArrayGet(pTableList->pTableList, index);
969,400✔
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) {
182!
2624
    return -1;
×
2625
  }
2626

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

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

2646
uint64_t tableListGetTableGroupId(const STableListInfo* pTableList, uint64_t tableUid) {
3,493,941✔
2647
  int32_t* slot = taosHashGet(pTableList->map, &tableUid, sizeof(tableUid));
3,493,941✔
2648
  if (slot == NULL) {
3,497,340!
2649
    return -1;
×
2650
  }
2651

2652
  STableKeyInfo* pKeyInfo = taosArrayGet(pTableList->pTableList, *slot);
3,497,340✔
2653

2654
  return pKeyInfo->groupId;
3,495,955✔
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) {
3,512✔
2659
  int32_t code = TSDB_CODE_SUCCESS;
3,512✔
2660
  int32_t lino = 0;
3,512✔
2661
  if (pTableList->map == NULL) {
3,512!
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};
3,512✔
2667
  void*         p = taosHashGet(pTableList->map, &uid, sizeof(uid));
3,512✔
2668
  if (p != NULL) {
3,511✔
2669
    qInfo("table:%" PRId64 " already in tableIdList, ignore it", uid);
3!
2670
    goto _end;
3✔
2671
  }
2672

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

2676
  int32_t slot = (int32_t)taosArrayGetSize(pTableList->pTableList) - 1;
3,506✔
2677
  code = taosHashPut(pTableList->map, &uid, sizeof(uid), &slot, sizeof(slot));
3,506✔
2678
  if (code != TSDB_CODE_SUCCESS) {
3,525!
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:
3,525✔
2685
  if (code != TSDB_CODE_SUCCESS) {
3,528!
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);
3,528✔
2689
  }
2690

2691
  return code;
3,528✔
2692
}
2693

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

2704
  if (ordinalGroupIndex < 0 || ordinalGroupIndex >= totalGroups) {
935,417!
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) {
935,417✔
2711
    *size = numOfTables;
934,481✔
2712
    *pKeyInfo = (*size == 0) ? NULL : taosArrayGet(pTableList->pTableList, 0);
934,481✔
2713
    return TSDB_CODE_SUCCESS;
934,025✔
2714
  } else if (totalGroups == numOfTables) {
936!
2715
    *size = 1;
951✔
2716
    *pKeyInfo = taosArrayGet(pTableList->pTableList, ordinalGroupIndex);
951✔
2717
    return TSDB_CODE_SUCCESS;
951✔
2718
  }
2719

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

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

2731
int32_t tableListGetOutputGroups(const STableListInfo* pTableList) { return pTableList->numOfOuputGroups; }
2,754,338✔
2732

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

2735
STableListInfo* tableListCreate() {
1,189,485✔
2736
  STableListInfo* pListInfo = taosMemoryCalloc(1, sizeof(STableListInfo));
1,189,485!
2737
  if (pListInfo == NULL) {
1,192,537!
2738
    return NULL;
×
2739
  }
2740

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

2747
  pListInfo->map = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
1,192,603✔
2748
  if (pListInfo->map == NULL) {
1,192,221!
2749
    goto _error;
×
2750
  }
2751

2752
  pListInfo->numOfOuputGroups = 1;
1,192,221✔
2753
  return pListInfo;
1,192,221✔
2754

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

2760
void tableListDestroy(STableListInfo* pTableListInfo) {
1,193,122✔
2761
  if (pTableListInfo == NULL) {
1,193,122✔
2762
    return;
220✔
2763
  }
2764

2765
  taosArrayDestroy(pTableListInfo->pTableList);
1,192,902✔
2766
  taosMemoryFreeClear(pTableListInfo->groupOffset);
1,193,121!
2767

2768
  taosHashCleanup(pTableListInfo->map);
1,193,121✔
2769
  taosHashCleanup(pTableListInfo->remainGroups);
1,193,158✔
2770
  pTableListInfo->pTableList = NULL;
1,193,181✔
2771
  pTableListInfo->map = NULL;
1,193,181✔
2772
  taosMemoryFree(pTableListInfo);
1,193,181!
2773
}
2774

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

2780
  taosArrayClear(pTableListInfo->pTableList);
2,502✔
2781
  taosHashClear(pTableListInfo->map);
2,501✔
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) {
80,318✔
2789
  STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;
80,318✔
2790
  STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;
80,318✔
2791

2792
  if (pInfo1->groupId == pInfo2->groupId) {
80,318✔
2793
    return 0;
168✔
2794
  } else {
2795
    return pInfo1->groupId < pInfo2->groupId ? -1 : 1;
80,150✔
2796
  }
2797
}
2798

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

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

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

2815
  int32_t start = 0;
867✔
2816
  void*   tmp = taosArrayPush(pList, &start);
867✔
2817
  if (!tmp) {
867!
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) {
2,759✔
2823
    pInfo = taosArrayGet(pTableListInfo->pTableList, i);
1,892✔
2824
    if (!pInfo) {
1,892✔
2825
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
1!
2826
      return terrno;
1✔
2827
    }
2828
    if (pInfo->groupId != gid) {
1,891✔
2829
      tmp = taosArrayPush(pList, &i);
1,727✔
2830
      if (!tmp) {
1,727!
2831
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
2832
        return terrno;
×
2833
      }
2834
      gid = pInfo->groupId;
1,727✔
2835
    }
2836
  }
2837

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

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

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

2854
  bool   groupByTbname = groupbyTbname(group);
1,038,970✔
2855
  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
1,038,826✔
2856
  if (!numOfTables) {
1,038,626!
2857
    return code;
×
2858
  }
2859
  if (group == NULL || groupByTbname) {
1,038,626✔
2860
    if (tsCountAlwaysReturnValue && QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode) &&
998,574✔
2861
        ((STableScanPhysiNode*)pScanNode)->needCountEmptyTable) {
762,266✔
2862
      pTableListInfo->remainGroups =
78,318✔
2863
          taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
78,327✔
2864
      if (pTableListInfo->remainGroups == NULL) {
78,318!
2865
        return terrno;
×
2866
      }
2867

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

2876
        int32_t tempRes = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId),
358,244✔
2877
                                      &(info->uid), sizeof(info->uid));
358,244✔
2878
        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
358,398!
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++) {
3,461,447✔
2885
        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
2,540,962✔
2886
        if (!info) {
2,541,458✔
2887
          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
258!
2888
          return terrno;
258✔
2889
        }
2890
        info->groupId = groupByTbname ? info->uid : 0;
2,541,200✔
2891
      }
2892
    }
2893

2894
    pTableListInfo->oneTableForEachGroup = groupByTbname;
998,944✔
2895
    if (numOfTables == 1 && pTableListInfo->idInfo.tableType == TSDB_CHILD_TABLE) {
998,944✔
2896
      pTableListInfo->oneTableForEachGroup = true;
232,051✔
2897
    }
2898

2899
    if (groupSort && groupByTbname) {
998,944✔
2900
      taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
19,289✔
2901
      pTableListInfo->numOfOuputGroups = numOfTables;
19,287✔
2902
    } else if (groupByTbname && pScanNode->groupOrderScan) {
979,655✔
2903
      pTableListInfo->numOfOuputGroups = numOfTables;
107✔
2904
    } else {
2905
      pTableListInfo->numOfOuputGroups = 1;
979,548✔
2906
    }
2907
  } else {
2908
    bool initRemainGroups = false;
40,052✔
2909
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {
40,052✔
2910
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;
39,480✔
2911
      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&
39,480✔
2912
          !(groupSort || pScanNode->groupOrderScan)) {
13,113!
2913
        initRemainGroups = true;
13,069✔
2914
      }
2915
    }
2916

2917
    code = getColInfoResultForGroupby(pHandle->vnode, group, pTableListInfo, digest, pAPI, initRemainGroups);
40,052✔
2918
    if (code != TSDB_CODE_SUCCESS) {
40,054✔
2919
      return code;
15✔
2920
    }
2921

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

2924
    if (groupSort || pScanNode->groupOrderScan) {
40,039✔
2925
      code = sortTableGroup(pTableListInfo);
858✔
2926
    }
2927
  }
2928

2929
  // add all table entry in the hash map
2930
  size_t size = taosArrayGetSize(pTableListInfo->pTableList);
1,038,990✔
2931
  for (int32_t i = 0; i < size; ++i) {
4,128,209✔
2932
    STableKeyInfo* p = taosArrayGet(pTableListInfo->pTableList, i);
3,087,537✔
2933
    if (!p) {
3,085,861!
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));
3,085,861✔
2938
    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
3,089,512!
2939
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
×
2940
      return tempRes;
×
2941
    }
2942
  }
2943

2944
  return code;
1,040,672✔
2945
}
2946

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

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

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

2966
  int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
1,102,742✔
2967

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

2973
  if (numOfTables == 0) {
1,103,390✔
2974
    qDebug("no table qualified for query, %s" PRIx64, idStr);
63,460✔
2975
    return TSDB_CODE_SUCCESS;
63,455✔
2976
  }
2977

2978
  code = buildGroupIdMapForAllTables(pTableListInfo, pHandle, pScanNode, pGroupTags, groupSort, digest,
1,039,930✔
2979
                                     &pTaskInfo->storageAPI);
2980
  if (code != TSDB_CODE_SUCCESS) {
1,040,259✔
2981
    return code;
15✔
2982
  }
2983

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

2987
  return TSDB_CODE_SUCCESS;
1,040,264✔
2988
}
2989

2990
char* getStreamOpName(uint16_t opType) {
1,694,609✔
2991
  switch (opType) {
1,694,609!
2992
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
66,747✔
2993
      return "stream scan";
66,747✔
2994
    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
27,208✔
2995
      return "project";
27,208✔
2996
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:
777,204✔
2997
      return "interval single";
777,204✔
2998
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL:
10,151✔
2999
      return "interval final";
10,151✔
3000
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
32,879✔
3001
      return "interval semi";
32,879✔
3002
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL:
331✔
3003
      return "interval mid";
331✔
3004
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
5,554✔
3005
      return "stream fill";
5,554✔
3006
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:
9,175✔
3007
      return "session single";
9,175✔
3008
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION:
1,218✔
3009
      return "session semi";
1,218✔
3010
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION:
978✔
3011
      return "session final";
978✔
3012
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:
6,791✔
3013
      return "state single";
6,791✔
3014
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
729,684✔
3015
      return "stream partitionby";
729,684✔
3016
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT:
830✔
3017
      return "stream event";
830✔
3018
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT:
9,836✔
3019
      return "stream count";
9,836✔
3020
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC:
9,444✔
3021
      return "stream interp";
9,444✔
3022
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL:
5,459✔
3023
      return "interval continue";
5,459✔
3024
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_INTERVAL:
155✔
3025
      return "interval continue semi";
155✔
3026
    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_INTERVAL:
102✔
3027
      return "interval continue final";
102✔
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";
863✔
3042
}
3043

3044
void printDataBlock(SSDataBlock* pBlock, const char* flag, const char* taskIdStr) {
813,859✔
3045
  if (!pBlock) {
813,859✔
3046
    qInfo("%s===stream===%s: Block is Null", taskIdStr, flag);
57!
3047
    return;
57✔
3048
  } else if (pBlock->info.rows == 0) {
813,802✔
3049
    qInfo("%s===stream===%s: Block is Empty. block type %d", taskIdStr, flag, pBlock->info.type);
19,818!
3050
    return;
19,857✔
3051
  }
3052
  if (qDebugFlag & DEBUG_DEBUG) {
793,984✔
3053
    char*   pBuf = NULL;
46,160✔
3054
    int32_t code = dumpBlockData(pBlock, flag, &pBuf, taskIdStr);
46,160✔
3055
    if (code == 0) {
46,160!
3056
      qInfo("%s", pBuf);
46,160!
3057
      taosMemoryFree(pBuf);
46,161!
3058
    }
3059
  }
3060
}
3061

3062
void printSpecDataBlock(SSDataBlock* pBlock, const char* flag, const char* opStr, const char* taskIdStr) {
817,582✔
3063
  if (!pBlock) {
817,582!
3064
    qDebug("%s===stream===%s %s: Block is Null", taskIdStr, flag, opStr);
×
3065
    return;
×
3066
  } else if (pBlock->info.rows == 0) {
817,582✔
3067
    qDebug("%s===stream===%s %s: Block is Empty. block type %d.skey:%" PRId64 ",ekey:%" PRId64 ",version%" PRId64,
12,634✔
3068
           taskIdStr, flag, opStr, pBlock->info.type, pBlock->info.window.skey, pBlock->info.window.ekey,
3069
           pBlock->info.version);
3070
    return;
12,635✔
3071
  }
3072
  if (qDebugFlag & DEBUG_DEBUG) {
804,948✔
3073
    char* pBuf = NULL;
58,169✔
3074
    char  flagBuf[64];
3075
    snprintf(flagBuf, sizeof(flagBuf), "%s %s", flag, opStr);
58,169✔
3076
    int32_t code = dumpBlockData(pBlock, flagBuf, &pBuf, taskIdStr);
58,169✔
3077
    if (code == 0) {
58,170!
3078
      qDebug("%s", pBuf);
58,170!
3079
      taosMemoryFree(pBuf);
58,171!
3080
    }
3081
  }
3082
}
3083

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

3086
void updateTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pWin, int64_t delta) {
117,873,421✔
3087
  int64_t* ts = (int64_t*)pColData->pData;
117,873,421✔
3088

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

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

3101
  for (int32_t i = 0; i < pSortGroupCols->size; ++i) {
53,742,869✔
3102
    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
32,856,505✔
3103
    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
32,856,505✔
3104
    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
32,856,505!
3105

3106
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
65,713,010!
3107
      if (isNull[i] != 1) return 1;
2,181,109✔
3108
    } else {
3109
      if (isNull[i] != 0) return 1;
30,675,396✔
3110
      const char* val = colDataGetData(pColInfoData, rowIndex);
30,661,176!
3111
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
30,661,176!
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)) {
30,661,176!
3116
        if (memcmp(p, val, varDataTLen(val)) != 0) return 1;
10,485,485✔
3117
        p += varDataTLen(val);
10,348,305✔
3118
      } else {
3119
        if (0 != memcmp(p, val, pCol->bytes)) return 1;
20,175,691✔
3120
        p += pCol->bytes;
19,843,479✔
3121
      }
3122
    }
3123
  }
3124
  if ((int32_t)(p - oldkeyBuf) != oldKeysLen) return 1;
20,886,364✔
3125
  return 0;
20,886,068✔
3126
}
3127

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

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

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

3141
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
1,916,712!
3142
      isNull[i] = 1;
34,530✔
3143
    } else {
3144
      isNull[i] = 0;
923,826✔
3145
      const char* val = colDataGetData(pColInfoData, rowIndex);
923,826!
3146
      if (pCol->type == TSDB_DATA_TYPE_JSON) {
923,826!
3147
        int32_t len = getJsonValueLen(val);
×
3148
        memcpy(p, val, len);
×
3149
        p += len;
×
3150
      } else if (IS_VAR_DATA_TYPE(pCol->type)) {
923,826✔
3151
        varDataCopy(p, val);
140,659✔
3152
        p += varDataTLen(val);
140,659✔
3153
      } else {
3154
        memcpy(p, val, pCol->bytes);
783,167✔
3155
        p += pCol->bytes;
783,167✔
3156
      }
3157
    }
3158
  }
3159
  return (int32_t)(p - keyBuf);
483,902✔
3160
}
3161

3162
uint64_t calcGroupId(char* pData, int32_t len) {
13,014,292✔
3163
  T_MD5_CTX context;
3164
  tMD5Init(&context);
13,014,292✔
3165
  tMD5Update(&context, (uint8_t*)pData, len);
12,984,776✔
3166
  tMD5Final(&context);
13,120,527✔
3167

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

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

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

3203
_end:
943✔
3204
  if (code != TSDB_CODE_SUCCESS) {
943!
3205
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
3206
  }
3207
  return code;
943✔
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