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

taosdata / TDengine / #4947

02 Feb 2026 09:27AM UTC coverage: 66.872% (-0.06%) from 66.932%
#4947

push

travis-ci

web-flow
enh: [6690002267] Optimize virtual table query with plenty of columns. (#34341)

527 of 634 new or added lines in 23 files covered. (83.12%)

3610 existing lines in 126 files now uncovered.

205539 of 307364 relevant lines covered (66.87%)

125933663.91 hits per line

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

69.5
/source/libs/executor/src/dataDispatcher.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 "dataSinkInt.h"
17
#include "dataSinkMgt.h"
18
#include "executorInt.h"
19
#include "planner.h"
20
#include "tcompression.h"
21
#include "tdatablock.h"
22
#include "tglobal.h"
23
#include "tqueue.h"
24

25
extern SDataSinkStat gDataSinkStat;
26

27
typedef struct SDataDispatchBuf {
28
  int32_t useSize;
29
  int32_t allocSize;
30
  char*   pData;
31
} SDataDispatchBuf;
32

33
typedef struct SDataCacheEntry {
34
  int32_t rawLen;
35
  int32_t dataLen;
36
  int32_t numOfRows;
37
  int32_t numOfCols;
38
  int8_t  compressed;
39
  char    data[];
40
} SDataCacheEntry;
41

42
typedef struct SDataDispatchHandle {
43
  SDataSinkHandle     sink;
44
  SDataSinkManager*   pManager;
45
  SDataBlockDescNode* pSchema;
46
  SDataSinkNode*      pSinkNode;
47
  STaosQueue*         pDataBlocks;
48
  SDataDispatchBuf    nextOutput;
49
  int32_t             outPutColCounts;
50
  int32_t             status;
51
  bool                queryEnd;
52
  uint64_t            useconds;
53
  uint64_t            cachedSize;
54
  uint64_t            flags;
55
  void*               pCompressBuf;
56
  int32_t             bufSize;
57
  bool                dynamicSchema;
58
  TdThreadMutex       mutex;
59
} SDataDispatchHandle;
60

61
static int32_t inputSafetyCheck(SDataDispatchHandle* pHandle, const SInputData* pInput)  {
514,488,110✔
62
  if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {
514,488,110✔
UNCOV
63
    return TSDB_CODE_SUCCESS;
×
64
  }
65
  if (pInput == NULL || pInput->pData == NULL || pInput->pData->info.rows <= 0) {
514,488,110✔
66
    qError("invalid input data");
34✔
UNCOV
67
    return TSDB_CODE_QRY_INVALID_INPUT;
×
68
  }
69
  SDataBlockDescNode* pSchema = pHandle->pSchema;
514,487,417✔
70
  if (pSchema == NULL || pSchema->totalRowSize != pInput->pData->info.rowSize) {
514,494,510✔
71
    qError("invalid schema");
35,650✔
UNCOV
72
    return TSDB_CODE_QRY_INVALID_INPUT;
×
73
  }
74

75
  if (pHandle->outPutColCounts > taosArrayGetSize(pInput->pData->pDataBlock)) {
514,459,314✔
UNCOV
76
    qError("invalid column number, schema:%d, input:%zu", pHandle->outPutColCounts, taosArrayGetSize(pInput->pData->pDataBlock));
×
77
    return TSDB_CODE_QRY_INVALID_INPUT;
×
78
  }
79

80
  SNode*  pNode;
81
  int32_t colNum = 0;
514,486,911✔
82
  FOREACH(pNode, pHandle->pSchema->pSlots) {
2,147,483,647✔
83
    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
2,147,483,647✔
84
    if (pSlotDesc->output) {
2,147,483,647✔
85
      SColumnInfoData* pColInfoData = taosArrayGet(pInput->pData->pDataBlock, colNum);
2,147,483,647✔
86
      if (pColInfoData == NULL) {
2,147,483,647✔
UNCOV
87
        return -1;
×
88
      }
89
      if (pColInfoData->info.bytes < 0) {
2,147,483,647✔
UNCOV
90
        qError("invalid column bytes, schema:%d, input:%d", pSlotDesc->dataType.bytes, pColInfoData->info.bytes);
×
91
        return TSDB_CODE_TSC_INTERNAL_ERROR;
×
92
      }
93
      if (!IS_VAR_DATA_TYPE(pColInfoData->info.type) &&
2,147,483,647✔
94
          TYPE_BYTES[pColInfoData->info.type] != pColInfoData->info.bytes) {
2,147,483,647✔
UNCOV
95
        qError("invalid column bytes, schema:%d, input:%d", TYPE_BYTES[pColInfoData->info.type],
×
96
               pColInfoData->info.bytes);
UNCOV
97
        return TSDB_CODE_TSC_INTERNAL_ERROR;
×
98
      }
99
      if (pColInfoData->info.type != pSlotDesc->dataType.type) {
2,147,483,647✔
UNCOV
100
        qError("invalid column type, schema:%d, input:%d", pSlotDesc->dataType.type, pColInfoData->info.type);
×
101
        return TSDB_CODE_QRY_INVALID_INPUT;
×
102
      }
103
      if (pColInfoData->info.bytes != pSlotDesc->dataType.bytes) {
2,147,483,647✔
UNCOV
104
        qError("invalid column bytes, schema:%d, input:%d", pSlotDesc->dataType.bytes, pColInfoData->info.bytes);
×
105
        return TSDB_CODE_QRY_INVALID_INPUT;
×
106
      }
107

108
      if (IS_INVALID_TYPE(pColInfoData->info.type)) {
2,147,483,647✔
109
        qError("invalid column type, type:%d", pColInfoData->info.type);
201✔
UNCOV
110
        return TSDB_CODE_TSC_INTERNAL_ERROR;
×
111
      }
112
      ++colNum;
2,147,483,647✔
113
    }
114
  }
115

116

117
  return TSDB_CODE_SUCCESS;
514,476,534✔
118
}
119

120
static int32_t inputSafetyCheckForDynamicSchema(const SInputData* pInput)  {
19,860,057✔
121
  if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {
19,860,057✔
NEW
122
    return TSDB_CODE_SUCCESS;
×
123
  }
124
  if (pInput == NULL || pInput->pData == NULL || pInput->pData->info.rows <= 0) {
19,860,057✔
NEW
125
    qError("invalid input data");
×
NEW
126
    return TSDB_CODE_QRY_INVALID_INPUT;
×
127
  }
128

129
  for (int32_t colNum = 0; colNum < taosArrayGetSize(pInput->pData->pDataBlock); colNum++) {
94,524,841✔
130
    SColumnInfoData* pColInfoData = taosArrayGet(pInput->pData->pDataBlock, colNum);
74,664,784✔
131
    if (pColInfoData == NULL) {
74,664,784✔
NEW
132
      return -1;
×
133
    }
134
    if (pColInfoData->info.bytes < 0) {
74,664,784✔
NEW
135
      qError("invalid column bytes, input:%d", pColInfoData->info.bytes);
×
NEW
136
      return TSDB_CODE_TSC_INTERNAL_ERROR;
×
137
    }
138
    if (!IS_VAR_DATA_TYPE(pColInfoData->info.type) &&
74,664,784✔
139
        TYPE_BYTES[pColInfoData->info.type] != pColInfoData->info.bytes) {
59,655,493✔
NEW
140
      qError("invalid column bytes, schema:%d, input:%d", TYPE_BYTES[pColInfoData->info.type],
×
141
             pColInfoData->info.bytes);
NEW
142
      return TSDB_CODE_TSC_INTERNAL_ERROR;
×
143
    }
144

145
    if (IS_INVALID_TYPE(pColInfoData->info.type)) {
74,664,784✔
NEW
146
      qError("invalid column type, type:%d", pColInfoData->info.type);
×
NEW
147
      return TSDB_CODE_TSC_INTERNAL_ERROR;
×
148
    }
149
  }
150

151
  return TSDB_CODE_SUCCESS;
19,860,057✔
152
}
153

154
// clang-format off
155
// data format:
156
// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+
157
// |SDataCacheEntry |  version         | total length | numOfRows    |     group id     | col1_schema | col2_schema | col3_schema... | column#1 length, column#2 length...| col1 bitmap | col1 data | col2 bitmap | col2 data |
158
// |                |  sizeof(int32_t) |sizeof(int32) | sizeof(int32)| sizeof(uint64_t) | (sizeof(int8_t)+sizeof(int32_t))*numOfCols | sizeof(int32_t) * numOfCols        | actual size |           |                         |
159
// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+
160
// The length of bitmap is decided by number of rows of this data block, and the length of each column data is
161
// recorded in the first segment, next to the struct header
162
// clang-format on
163
static int32_t toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) {
534,346,573✔
164
  int32_t numOfCols = 0;
534,346,573✔
165
  SNode*  pNode;
166
  int32_t code = TSDB_CODE_SUCCESS;
534,346,573✔
167

168
  if (pHandle->dynamicSchema) {
534,346,573✔
169
    code = inputSafetyCheckForDynamicSchema(pInput);
19,860,057✔
170
    if (code) {
19,860,057✔
NEW
171
      qError("failed to check input data for dynamic schema, code:%d", code);
×
NEW
172
      return code;
×
173
    }
174
    numOfCols = taosArrayGetSize(pInput->pData->pDataBlock);
19,860,057✔
175
  } else {
176
    code = inputSafetyCheck(pHandle, pInput);
514,494,360✔
177
    if (code) {
514,479,572✔
NEW
178
      qError("failed to check input data, code:%d", code);
×
NEW
179
      return code;
×
180
    }
181

182
    FOREACH(pNode, pHandle->pSchema->pSlots) {
2,147,483,647✔
183
      SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
2,147,483,647✔
184
      if (pSlotDesc->output) {
2,147,483,647✔
185
        ++numOfCols;
2,147,483,647✔
186
      }
187
    }
188
  }
189

190
  SDataCacheEntry* pEntry = (SDataCacheEntry*)pBuf->pData;
534,297,829✔
191
  pEntry->compressed = 0;
534,303,124✔
192
  pEntry->numOfRows = pInput->pData->info.rows;
534,293,238✔
193
  pEntry->numOfCols = numOfCols;
534,320,627✔
194
  pEntry->dataLen = 0;
534,332,840✔
195
  pEntry->rawLen = 0;
534,313,694✔
196

197
  pBuf->useSize = sizeof(SDataCacheEntry);
534,274,600✔
198

199
  {
200
    // allocate additional 8 bytes to avoid invalid write if compress failed to reduce the size
201
    size_t dataEncodeBufSize = pBuf->allocSize + 8;
534,299,034✔
202
    if ((pBuf->allocSize > tsCompressMsgSize) && (tsCompressMsgSize > 0) && pHandle->pManager->cfg.compress) {
534,313,682✔
UNCOV
203
      if (pHandle->pCompressBuf == NULL) {
×
UNCOV
204
        pHandle->pCompressBuf = taosMemoryMalloc(dataEncodeBufSize);
×
UNCOV
205
        if (NULL == pHandle->pCompressBuf) {
×
UNCOV
206
          QRY_RET(terrno);
×
207
        }
UNCOV
208
        pHandle->bufSize = dataEncodeBufSize;
×
209
      } else {
UNCOV
210
        if (pHandle->bufSize < dataEncodeBufSize) {
×
UNCOV
211
          pHandle->bufSize = dataEncodeBufSize;
×
UNCOV
212
          void* p = taosMemoryRealloc(pHandle->pCompressBuf, pHandle->bufSize);
×
UNCOV
213
          if (p != NULL) {
×
UNCOV
214
            pHandle->pCompressBuf = p;
×
215
          } else {
216
            qError("failed to prepare compress buf:%d, code: %x", pHandle->bufSize, terrno);
×
217
            return terrno;
×
218
          }
219
        }
220
      }
221

UNCOV
222
      int32_t dataLen = blockEncode(pInput->pData, pHandle->pCompressBuf, dataEncodeBufSize, numOfCols);
×
223
      if(dataLen < 0) {
×
224
        qError("failed to encode data block, code: %d", dataLen);
×
UNCOV
225
        return terrno;
×
226
      }
227
      int32_t len =
UNCOV
228
          tsCompressString(pHandle->pCompressBuf, dataLen, 1, pEntry->data, pBuf->allocSize, ONE_STAGE_COMP, NULL, 0);
×
UNCOV
229
      if (len < dataLen) {
×
UNCOV
230
        pEntry->compressed = 1;
×
UNCOV
231
        pEntry->dataLen = len;
×
UNCOV
232
        pEntry->rawLen = dataLen;
×
233
      } else {  // no need to compress data
UNCOV
234
        pEntry->compressed = 0;
×
UNCOV
235
        pEntry->dataLen = dataLen;
×
UNCOV
236
        pEntry->rawLen = dataLen;
×
UNCOV
237
        TAOS_MEMCPY(pEntry->data, pHandle->pCompressBuf, dataLen);
×
238
      }
239
    } else {
240
      pEntry->dataLen =
534,316,853✔
241
          pHandle->dynamicSchema
534,340,460✔
242
              ? blockEncodeInternal(pInput->pData, pEntry->data,  pBuf->allocSize, numOfCols)
19,860,057✔
243
              : blockEncode(pInput->pData, pEntry->data,  pBuf->allocSize, numOfCols);
534,277,392✔
244
      if(pEntry->dataLen < 0) {
534,324,385✔
UNCOV
245
        qError("failed to encode data block, code: %d", pEntry->dataLen);
×
UNCOV
246
        return terrno;
×
247
      }
248
      pEntry->rawLen = pEntry->dataLen;
534,287,165✔
249
    }
250
  }
251

252
  pBuf->useSize += pEntry->dataLen;
534,302,895✔
253

254
  (void)atomic_add_fetch_64(&pHandle->cachedSize, pEntry->dataLen);
534,321,975✔
255
  (void)atomic_add_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);
534,339,380✔
256

257
  return TSDB_CODE_SUCCESS;
534,262,797✔
258
}
259

260
static int32_t allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput, SDataDispatchBuf* pBuf) {
534,352,754✔
261
  /*
262
    uint32_t capacity = pDispatcher->pManager->cfg.maxDataBlockNumPerQuery;
263
    if (taosQueueItemSize(pDispatcher->pDataBlocks) > capacity) {
264
      qError("SinkNode queue is full, no capacity, max:%d, current:%d, no capacity", capacity,
265
             taosQueueItemSize(pDispatcher->pDataBlocks));
266
      return false;
267
    }
268
  */
269

270
  pBuf->allocSize = sizeof(SDataCacheEntry) + (pDispatcher->dynamicSchema
1,068,691,424✔
271
                                                   ? blockGetInternalEncodeSize(pInput->pData)
19,860,057✔
272
                                                   : blockGetEncodeSize(pInput->pData));
514,493,962✔
273

274
  pBuf->pData = taosMemoryMalloc(pBuf->allocSize);
534,344,562✔
275
  if (pBuf->pData == NULL) {
534,322,869✔
UNCOV
276
    qError("SinkNode failed to malloc memory, size:%d, code:%x", pBuf->allocSize, terrno);
×
UNCOV
277
    return terrno;
×
278
  }
279

280
  return TSDB_CODE_SUCCESS;
534,340,299✔
281
}
282

283
static int32_t updateStatus(SDataDispatchHandle* pDispatcher) {
1,060,675,819✔
284
  (void)taosThreadMutexLock(&pDispatcher->mutex);
1,060,675,819✔
285
  int32_t blockNums = taosQueueItemSize(pDispatcher->pDataBlocks);
1,060,789,603✔
286
  int32_t status =
1,060,740,586✔
287
      (0 == blockNums ? DS_BUF_EMPTY
288
                      : (blockNums < pDispatcher->pManager->cfg.maxDataBlockNumPerQuery ? DS_BUF_LOW : DS_BUF_FULL));
1,060,723,905✔
289
  pDispatcher->status = status;
1,060,740,586✔
290
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
1,060,751,171✔
291
  return status;
1,060,793,880✔
292
}
293

UNCOV
294
static int32_t getStatus(SDataDispatchHandle* pDispatcher) {
×
295
  (void)taosThreadMutexLock(&pDispatcher->mutex);
×
296
  int32_t status = pDispatcher->status;
×
UNCOV
297
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
×
UNCOV
298
  return status;
×
299
}
300

301
static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
534,358,454✔
302
  int32_t              code = 0;
534,358,454✔
303
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
534,358,454✔
304
  SDataDispatchBuf*    pBuf = NULL;
534,358,454✔
305

306
  code = taosAllocateQitem(sizeof(SDataDispatchBuf), DEF_QITEM, 0, (void**)&pBuf);
534,361,564✔
307
  if (code) {
534,356,994✔
UNCOV
308
    return code;
×
309
  }
310

311
  code = allocBuf(pDispatcher, pInput, pBuf);
534,356,994✔
312
  if (code) {
534,350,383✔
UNCOV
313
    taosFreeQitem(pBuf);
×
UNCOV
314
    return code;
×
315
  }
316

317
  QRY_ERR_JRET(toDataCacheEntry(pDispatcher, pInput, pBuf));
534,350,383✔
318
  QRY_ERR_JRET(taosWriteQitem(pDispatcher->pDataBlocks, pBuf));
534,262,463✔
319

320
  int32_t status = updateStatus(pDispatcher);
534,305,655✔
321
  *pContinue = (status == DS_BUF_LOW || status == DS_BUF_EMPTY) && !(pDispatcher->flags & DS_FLAG_PROCESS_ONE_BLOCK);
534,335,052✔
322
  return TSDB_CODE_SUCCESS;
534,335,853✔
323

UNCOV
324
_return:
×
325

326
  taosMemoryFreeClear(pBuf->pData);
×
327
  taosFreeQitem(pBuf);
×
UNCOV
328
  return code;
×
329
}
330

331
static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {
263,974,180✔
332
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
263,974,180✔
333
  (void)taosThreadMutexLock(&pDispatcher->mutex);
263,974,180✔
334
  pDispatcher->queryEnd = true;
263,984,511✔
335
  pDispatcher->useconds = useconds;
263,962,484✔
336
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
263,958,953✔
337
}
263,978,938✔
338

339
static void resetDispatcher(struct SDataSinkHandle* pHandle) {
31,395,434✔
340
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
31,395,434✔
341
  (void)taosThreadMutexLock(&pDispatcher->mutex);
31,395,434✔
342
  pDispatcher->queryEnd = false;
31,400,597✔
343
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
31,400,597✔
344
}
31,399,082✔
345

346
static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRowLen, bool* pQueryEnd) {
620,283,962✔
347
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
620,283,962✔
348
  if (taosQueueEmpty(pDispatcher->pDataBlocks)) {
620,283,962✔
349
    *pQueryEnd = pDispatcher->queryEnd;
93,929,670✔
350
    *pLen = 0;
93,930,298✔
351
    return;
93,929,848✔
352
  }
353

354
  SDataDispatchBuf* pBuf = NULL;
526,387,754✔
355
  taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);
526,401,844✔
356
  if (pBuf != NULL) {
526,404,176✔
357
    TAOS_MEMCPY(&pDispatcher->nextOutput, pBuf, sizeof(SDataDispatchBuf));
526,434,220✔
358
    taosFreeQitem(pBuf);
526,397,729✔
359
  }
360

361
  SDataCacheEntry* pEntry = (SDataCacheEntry*)pDispatcher->nextOutput.pData;
526,340,045✔
362
  *pLen = pEntry->dataLen;
526,347,732✔
363
  *pRowLen = pEntry->rawLen;
526,417,981✔
364

365
  *pQueryEnd = pDispatcher->queryEnd;
526,420,231✔
366
  qDebug("got data len %" PRId64 ", row num %d in sink", *pLen,
526,412,242✔
367
         ((SDataCacheEntry*)(pDispatcher->nextOutput.pData))->numOfRows);
368
}
369

370
static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
582,598,501✔
371
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
582,598,501✔
372
  if (NULL == pDispatcher->nextOutput.pData) {
582,598,501✔
373
    if (!pDispatcher->queryEnd) {
56,204,127✔
374
      qError("empty res while query not end in data dispatcher");
×
UNCOV
375
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
×
376
    }
377

378
    pOutput->useconds = pDispatcher->useconds;
56,203,995✔
379
    pOutput->precision = pDispatcher->pSchema->precision;
56,203,817✔
380
    pOutput->bufStatus = DS_BUF_EMPTY;
56,206,228✔
381
    pOutput->queryEnd = pDispatcher->queryEnd;
56,206,228✔
382
    return TSDB_CODE_SUCCESS;
56,206,639✔
383
  }
384

385
  SDataCacheEntry* pEntry = (SDataCacheEntry*)(pDispatcher->nextOutput.pData);
526,349,842✔
386
  TAOS_MEMCPY(pOutput->pData, pEntry->data, pEntry->dataLen);
526,401,359✔
387
  pOutput->numOfRows = pEntry->numOfRows;
526,407,459✔
388
  pOutput->numOfCols = pEntry->numOfCols;
526,306,659✔
389
  pOutput->compressed = pEntry->compressed;
526,378,943✔
390

391
  (void)atomic_sub_fetch_64(&pDispatcher->cachedSize, pEntry->dataLen);
526,379,849✔
392
  (void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);
526,413,458✔
393

394
  taosMemoryFreeClear(pDispatcher->nextOutput.pData);  // todo persistent
526,433,958✔
395
  pOutput->bufStatus = updateStatus(pDispatcher);
526,392,313✔
396
  
397
  (void)taosThreadMutexLock(&pDispatcher->mutex);
526,405,034✔
398
  pOutput->queryEnd = pDispatcher->queryEnd;
526,433,944✔
399
  pOutput->useconds = pDispatcher->useconds;
526,393,031✔
400
  pOutput->precision = pDispatcher->pSchema->precision;
526,382,985✔
401
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
526,409,067✔
402

403
  return TSDB_CODE_SUCCESS;
526,436,156✔
404
}
405

406
static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
265,188,094✔
407
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
265,188,094✔
408
  (void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pDispatcher->cachedSize);
265,188,094✔
409
  taosMemoryFreeClear(pDispatcher->nextOutput.pData);
265,176,340✔
410
  nodesDestroyNode((SNode*)pDispatcher->pSinkNode);
265,161,471✔
411

412
  while (!taosQueueEmpty(pDispatcher->pDataBlocks)) {
273,046,227✔
413
    SDataDispatchBuf* pBuf = NULL;
7,923,929✔
414
    taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);
7,923,929✔
415
    if (pBuf != NULL) {
7,923,929✔
416
      taosMemoryFreeClear(pBuf->pData);
7,923,929✔
417
      taosFreeQitem(pBuf);
7,923,929✔
418
    }
419
  }
420

421
  taosCloseQueue(pDispatcher->pDataBlocks);
265,153,003✔
422
  taosMemoryFreeClear(pDispatcher->pCompressBuf);
265,089,983✔
423
  pDispatcher->bufSize = 0;
265,099,870✔
424

425
  (void)taosThreadMutexDestroy(&pDispatcher->mutex);
265,162,839✔
426
  taosMemoryFree(pDispatcher->pManager);
265,070,473✔
427
  return TSDB_CODE_SUCCESS;
265,008,460✔
428
}
429

UNCOV
430
static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {
×
UNCOV
431
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
×
432

UNCOV
433
  *size = atomic_load_64(&pDispatcher->cachedSize);
×
UNCOV
434
  return TSDB_CODE_SUCCESS;
×
435
}
436

437
static int32_t getSinkFlags(struct SDataSinkHandle* pHandle, uint64_t* pFlags) {
308,315,495✔
438
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
308,315,495✔
439

440
  *pFlags = atomic_load_64(&pDispatcher->flags);
308,315,495✔
441
  return TSDB_CODE_SUCCESS;
308,291,881✔
442
}
443

444
static int32_t blockDescNodeCheck(SDataBlockDescNode* pInputDataBlockDesc)  {
264,995,646✔
445
  if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {
264,995,646✔
UNCOV
446
    return TSDB_CODE_SUCCESS;
×
447
  }
448

449
  if (pInputDataBlockDesc == NULL) {
264,995,646✔
UNCOV
450
    qError("invalid schema");
×
UNCOV
451
    return TSDB_CODE_QRY_INVALID_INPUT;
×
452
  }
453

454
  SNode*  pNode;
455
  int32_t realOutputRowSize = 0;
264,995,646✔
456
  FOREACH(pNode, pInputDataBlockDesc->pSlots) {
1,588,360,124✔
457
    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
1,332,188,157✔
458
    if (pSlotDesc->output) {
1,332,188,157✔
459
      realOutputRowSize += pSlotDesc->dataType.bytes;
1,323,025,945✔
460
    } else {
461
      // Slots must be sorted, and slots with 'output' set to true must come first
462
      break;
8,663,597✔
463
    }
464
  }
465
  if (realOutputRowSize !=  pInputDataBlockDesc->outputRowSize) {
264,839,519✔
UNCOV
466
    qError("invalid schema, realOutputRowSize:%d, outputRowSize:%d", realOutputRowSize, pInputDataBlockDesc->outputRowSize);
×
UNCOV
467
    return TSDB_CODE_QRY_INVALID_INPUT;
×
468
  }
469
  return TSDB_CODE_SUCCESS;
264,960,441✔
470
}
471

472
int32_t getOutputColCounts(SDataBlockDescNode* pInputDataBlockDesc) {
264,902,887✔
473
  if (pInputDataBlockDesc == NULL) {
264,902,887✔
UNCOV
474
    qError("invalid schema");
×
UNCOV
475
    return 0;
×
476
  }
477
  SNode*  pNode;
478
  int32_t numOfCols = 0;
264,902,887✔
479
  FOREACH(pNode, pInputDataBlockDesc->pSlots) {
1,587,458,959✔
480
    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
1,331,491,064✔
481
    if (pSlotDesc->output) {
1,331,491,064✔
482
      ++numOfCols;
1,322,556,072✔
483
    } else {
484
      // Slots must be sorted, and slots with 'output' set to true must come first
485
      break;
8,523,786✔
486
    }
487
  }
488
  return numOfCols;
264,630,785✔
489
}
490

491
int32_t createDataDispatcher(SDataSinkManager* pManager, SDataSinkNode** ppDataSink, DataSinkHandle* pHandle, bool processOneBlock) {
264,982,783✔
492
  int32_t code;
493
  SDataSinkNode* pDataSink = *ppDataSink;
264,982,783✔
494
  code = blockDescNodeCheck(pDataSink->pInputDataBlockDesc);
265,063,221✔
495
  if (code) {
264,904,507✔
496
    qError("failed to check input data block desc, code:%d", code);
×
UNCOV
497
    goto _return;
×
498
  }
499

500
  SDataDispatcherNode* pDispatherNode = (SDataDispatcherNode*)pDataSink;
264,904,507✔
501
  SDataDispatchHandle* dispatcher = taosMemoryCalloc(1, sizeof(SDataDispatchHandle));
264,904,507✔
502
  if (NULL == dispatcher) {
264,899,235✔
UNCOV
503
    goto _return;
×
504
  }
505

506
  dispatcher->sink.fPut = putDataBlock;
264,899,235✔
507
  dispatcher->sink.fEndPut = endPut;
264,931,014✔
508
  dispatcher->sink.fReset = resetDispatcher;
264,922,759✔
509
  dispatcher->sink.fGetLen = getDataLength;
264,986,145✔
510
  dispatcher->sink.fGetData = getDataBlock;
265,054,604✔
511
  dispatcher->sink.fDestroy = destroyDataSinker;
264,818,704✔
512
  dispatcher->sink.fGetCacheSize = getCacheSize;
264,923,900✔
513
  dispatcher->sink.fGetFlags = getSinkFlags;
264,965,817✔
514
  dispatcher->pManager = pManager;
265,051,161✔
515
  pManager = NULL;
264,932,958✔
516
  dispatcher->dynamicSchema = pDispatherNode->dynamicSchema;
264,932,958✔
517
  dispatcher->pSchema = pDataSink->pInputDataBlockDesc;
264,975,162✔
518
  dispatcher->pSinkNode = pDataSink;
264,925,771✔
519
  *ppDataSink = NULL;
264,813,826✔
520
  dispatcher->outPutColCounts = getOutputColCounts(dispatcher->pSchema);
264,585,900✔
521
  dispatcher->status = DS_BUF_EMPTY;
264,659,568✔
522
  dispatcher->queryEnd = false;
264,727,257✔
523
  code = taosOpenQueue(&dispatcher->pDataBlocks);
264,897,412✔
524
  if (code) {
264,886,677✔
UNCOV
525
    terrno = code;
×
526
    goto _return;
×
527
  }
528
  code = taosThreadMutexInit(&dispatcher->mutex, NULL);
264,886,677✔
529
  if (code) {
264,840,616✔
UNCOV
530
    terrno = code;
×
UNCOV
531
    goto _return;
×
532
  }
533

534
  dispatcher->flags = DS_FLAG_USE_MEMPOOL;
264,840,616✔
535
  if (processOneBlock) {
264,844,573✔
536
    dispatcher->flags |= DS_FLAG_PROCESS_ONE_BLOCK;
4,450,835✔
537
  }
538

539
  *pHandle = dispatcher;
264,845,690✔
540
  return TSDB_CODE_SUCCESS;
264,947,701✔
541

UNCOV
542
_return:
×
543

UNCOV
544
  taosMemoryFree(pManager);
×
545
  
UNCOV
546
  if (dispatcher) {
×
UNCOV
547
    dsDestroyDataSinker(dispatcher);
×
548
  }
549

UNCOV
550
  nodesDestroyNode((SNode *)*ppDataSink);
×
UNCOV
551
  *ppDataSink = NULL;
×
552
  
UNCOV
553
  return terrno;
×
554
}
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