#4986

Committed 15 Mar 2026 08:32AM UTC coverage: 37.305% (-31.3%) from 68.601%

Build # #4986

Build Type

push

travis-ci

Committed by

tomchon

Commit Message

test: keep docs and unit test

Coverage Stats

125478 of 336361 relevant lines covered (37.3%)

1134847.06 hits per line

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

0.0

/source/libs/executor/src/dataDispatcher.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "dataSinkInt.h"
#include "dataSinkMgt.h"
#include "executorInt.h"
#include "planner.h"
#include "tcompression.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "tqueue.h"

extern SDataSinkStat gDataSinkStat;

typedef struct SDataDispatchBuf {
  int32_t useSize;
  int32_t allocSize;
  char*   pData;
} SDataDispatchBuf;

typedef struct SDataCacheEntry {
  int32_t rawLen;
  int32_t dataLen;
  int32_t numOfRows;
  int32_t numOfCols;
  int8_t  compressed;
  char    data[];
} SDataCacheEntry;

typedef struct SDataDispatchHandle {
  SDataSinkHandle     sink;
  SDataSinkManager*   pManager;
  SDataBlockDescNode* pSchema;
  SDataSinkNode*      pSinkNode;
  STaosQueue*         pDataBlocks;
  SDataDispatchBuf    nextOutput;
  int32_t             outPutColCounts;
  int32_t             status;
  bool                queryEnd;
  uint64_t            useconds;
  uint64_t            cachedSize;
  uint64_t            flags;
  void*               pCompressBuf;
  int32_t             bufSize;
  bool                dynamicSchema;
  TdThreadMutex       mutex;
} SDataDispatchHandle;

static int32_t inputSafetyCheck(SDataDispatchHandle* pHandle, const SInputData* pInput)  {
  if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {
    return TSDB_CODE_SUCCESS;
  }
  if (pInput == NULL || pInput->pData == NULL || pInput->pData->info.rows <= 0) {
    qError("invalid input data");
    return TSDB_CODE_QRY_INVALID_INPUT;
  }
  SDataBlockDescNode* pSchema = pHandle->pSchema;
  if (pSchema == NULL || pSchema->totalRowSize != pInput->pData->info.rowSize) {
    qError("invalid schema");
    return TSDB_CODE_QRY_INVALID_INPUT;
  }

  if (pHandle->outPutColCounts > taosArrayGetSize(pInput->pData->pDataBlock)) {
    qError("invalid column number, schema:%d, input:%zu", pHandle->outPutColCounts, taosArrayGetSize(pInput->pData->pDataBlock));
    return TSDB_CODE_QRY_INVALID_INPUT;
  }

  SNode*  pNode;
  int32_t colNum = 0;
  FOREACH(pNode, pHandle->pSchema->pSlots) {
    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
    if (pSlotDesc->output) {
      SColumnInfoData* pColInfoData = taosArrayGet(pInput->pData->pDataBlock, colNum);
      if (pColInfoData == NULL) {
        return -1;
      }
      if (pColInfoData->info.bytes < 0) {
        qError("invalid column bytes, schema:%d, input:%d", pSlotDesc->dataType.bytes, pColInfoData->info.bytes);
        return TSDB_CODE_TSC_INTERNAL_ERROR;
      }
      if (!IS_VAR_DATA_TYPE(pColInfoData->info.type) &&
          TYPE_BYTES[pColInfoData->info.type] != pColInfoData->info.bytes) {
        qError("invalid column bytes, schema:%d, input:%d", TYPE_BYTES[pColInfoData->info.type],
               pColInfoData->info.bytes);
        return TSDB_CODE_TSC_INTERNAL_ERROR;
      }
      if (pColInfoData->info.type != pSlotDesc->dataType.type) {
        qError("invalid column type, schema:%d, input:%d", pSlotDesc->dataType.type, pColInfoData->info.type);
        return TSDB_CODE_QRY_INVALID_INPUT;
      }
      if (pColInfoData->info.bytes != pSlotDesc->dataType.bytes) {
        qError("invalid column bytes, schema:%d, input:%d", pSlotDesc->dataType.bytes, pColInfoData->info.bytes);
        return TSDB_CODE_QRY_INVALID_INPUT;
      }

      if (IS_INVALID_TYPE(pColInfoData->info.type)) {
        qError("invalid column type, type:%d", pColInfoData->info.type);
        return TSDB_CODE_TSC_INTERNAL_ERROR;
      }
      ++colNum;
    }
  }


  return TSDB_CODE_SUCCESS;
}

static int32_t inputSafetyCheckForDynamicSchema(const SInputData* pInput)  {
  if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {
    return TSDB_CODE_SUCCESS;
  }
  if (pInput == NULL || pInput->pData == NULL || pInput->pData->info.rows <= 0) {
    qError("invalid input data");
    return TSDB_CODE_QRY_INVALID_INPUT;
  }

  for (int32_t colNum = 0; colNum < taosArrayGetSize(pInput->pData->pDataBlock); colNum++) {
    SColumnInfoData* pColInfoData = taosArrayGet(pInput->pData->pDataBlock, colNum);
    if (pColInfoData == NULL) {
      return -1;
    }
    if (pColInfoData->info.bytes < 0) {
      qError("invalid column bytes, input:%d", pColInfoData->info.bytes);
      return TSDB_CODE_TSC_INTERNAL_ERROR;
    }
    if (!IS_VAR_DATA_TYPE(pColInfoData->info.type) &&
        TYPE_BYTES[pColInfoData->info.type] != pColInfoData->info.bytes) {
      qError("invalid column bytes, schema:%d, input:%d", TYPE_BYTES[pColInfoData->info.type],
             pColInfoData->info.bytes);
      return TSDB_CODE_TSC_INTERNAL_ERROR;
    }

    if (IS_INVALID_TYPE(pColInfoData->info.type)) {
      qError("invalid column type, type:%d", pColInfoData->info.type);
      return TSDB_CODE_TSC_INTERNAL_ERROR;
    }
  }

  return TSDB_CODE_SUCCESS;
}

// clang-format off
// data format:
// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+
// |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 |
// |                |  sizeof(int32_t) |sizeof(int32) | sizeof(int32)| sizeof(uint64_t) | (sizeof(int8_t)+sizeof(int32_t))*numOfCols | sizeof(int32_t) * numOfCols        | actual size |           |                         |
// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+
// The length of bitmap is decided by number of rows of this data block, and the length of each column data is
// recorded in the first segment, next to the struct header
// clang-format on
static int32_t toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) {
  int32_t numOfCols = 0;
  SNode*  pNode;
  int32_t code = TSDB_CODE_SUCCESS;

  if (pHandle->dynamicSchema) {
    code = inputSafetyCheckForDynamicSchema(pInput);
    if (code) {
      qError("failed to check input data for dynamic schema, code:%d", code);
      return code;
    }
    numOfCols = taosArrayGetSize(pInput->pData->pDataBlock);
  } else {
    code = inputSafetyCheck(pHandle, pInput);
    if (code) {
      qError("failed to check input data, code:%d", code);
      return code;
    }

    FOREACH(pNode, pHandle->pSchema->pSlots) {
      SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
      if (pSlotDesc->output) {
        ++numOfCols;
      }
    }
  }

  SDataCacheEntry* pEntry = (SDataCacheEntry*)pBuf->pData;
  pEntry->compressed = 0;
  pEntry->numOfRows = pInput->pData->info.rows;
  pEntry->numOfCols = numOfCols;
  pEntry->dataLen = 0;
  pEntry->rawLen = 0;

  pBuf->useSize = sizeof(SDataCacheEntry);

  {
    // allocate additional 8 bytes to avoid invalid write if compress failed to reduce the size
    size_t dataEncodeBufSize = pBuf->allocSize + 8;
    if ((pBuf->allocSize > tsCompressMsgSize) && (tsCompressMsgSize > 0) && pHandle->pManager->cfg.compress) {
      if (pHandle->pCompressBuf == NULL) {
        pHandle->pCompressBuf = taosMemoryMalloc(dataEncodeBufSize);
        if (NULL == pHandle->pCompressBuf) {
          QRY_RET(terrno);
        }
        pHandle->bufSize = dataEncodeBufSize;
      } else {
        if (pHandle->bufSize < dataEncodeBufSize) {
          pHandle->bufSize = dataEncodeBufSize;
          void* p = taosMemoryRealloc(pHandle->pCompressBuf, pHandle->bufSize);
          if (p != NULL) {
            pHandle->pCompressBuf = p;
          } else {
            qError("failed to prepare compress buf:%d, code: %x", pHandle->bufSize, terrno);
            return terrno;
          }
        }
      }

      int32_t dataLen = blockEncode(pInput->pData, pHandle->pCompressBuf, dataEncodeBufSize, numOfCols);
      if(dataLen < 0) {
        qError("failed to encode data block, code: %d", dataLen);
        return terrno;
      }
      int32_t len =
          tsCompressString(pHandle->pCompressBuf, dataLen, 1, pEntry->data, pBuf->allocSize, ONE_STAGE_COMP, NULL, 0);
      if (len < dataLen) {
        pEntry->compressed = 1;
        pEntry->dataLen = len;
        pEntry->rawLen = dataLen;
      } else {  // no need to compress data
        pEntry->compressed = 0;
        pEntry->dataLen = dataLen;
        pEntry->rawLen = dataLen;
        TAOS_MEMCPY(pEntry->data, pHandle->pCompressBuf, dataLen);
      }
    } else {
      pEntry->dataLen =
          pHandle->dynamicSchema
              ? blockEncodeInternal(pInput->pData, pEntry->data,  pBuf->allocSize, numOfCols)
              : blockEncode(pInput->pData, pEntry->data,  pBuf->allocSize, numOfCols);
      if(pEntry->dataLen < 0) {
        qError("failed to encode data block, code: %d", pEntry->dataLen);
        return terrno;
      }
      pEntry->rawLen = pEntry->dataLen;
    }
  }

  pBuf->useSize += pEntry->dataLen;

  (void)atomic_add_fetch_64(&pHandle->cachedSize, pEntry->dataLen);
  (void)atomic_add_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);

  return TSDB_CODE_SUCCESS;
}

static int32_t allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput, SDataDispatchBuf* pBuf) {
  /*
    uint32_t capacity = pDispatcher->pManager->cfg.maxDataBlockNumPerQuery;
    if (taosQueueItemSize(pDispatcher->pDataBlocks) > capacity) {
      qError("SinkNode queue is full, no capacity, max:%d, current:%d, no capacity", capacity,
             taosQueueItemSize(pDispatcher->pDataBlocks));
      return false;
    }
  */

  pBuf->allocSize = sizeof(SDataCacheEntry) + (pDispatcher->dynamicSchema
                                                   ? blockGetInternalEncodeSize(pInput->pData)
                                                   : blockGetEncodeSize(pInput->pData));

  pBuf->pData = taosMemoryMalloc(pBuf->allocSize);
  if (pBuf->pData == NULL) {
    qError("SinkNode failed to malloc memory, size:%d, code:%x", pBuf->allocSize, terrno);
    return terrno;
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t updateStatus(SDataDispatchHandle* pDispatcher) {
  (void)taosThreadMutexLock(&pDispatcher->mutex);
  int32_t blockNums = taosQueueItemSize(pDispatcher->pDataBlocks);
  int32_t status =
      (0 == blockNums ? DS_BUF_EMPTY
                      : (blockNums < pDispatcher->pManager->cfg.maxDataBlockNumPerQuery ? DS_BUF_LOW : DS_BUF_FULL));
  pDispatcher->status = status;
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
  return status;
}

static int32_t getStatus(SDataDispatchHandle* pDispatcher) {
  (void)taosThreadMutexLock(&pDispatcher->mutex);
  int32_t status = pDispatcher->status;
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
  return status;
}

static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
  int32_t              code = 0;
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
  SDataDispatchBuf*    pBuf = NULL;

  code = taosAllocateQitem(sizeof(SDataDispatchBuf), DEF_QITEM, 0, (void**)&pBuf);
  if (code) {
    return code;
  }

  code = allocBuf(pDispatcher, pInput, pBuf);
  if (code) {
    taosFreeQitem(pBuf);
    return code;
  }

  QRY_ERR_JRET(toDataCacheEntry(pDispatcher, pInput, pBuf));
  QRY_ERR_JRET(taosWriteQitem(pDispatcher->pDataBlocks, pBuf));

  int32_t status = updateStatus(pDispatcher);
  *pContinue = (status == DS_BUF_LOW || status == DS_BUF_EMPTY) && !(pDispatcher->flags & DS_FLAG_PROCESS_ONE_BLOCK);
  return TSDB_CODE_SUCCESS;

_return:

  taosMemoryFreeClear(pBuf->pData);
  taosFreeQitem(pBuf);
  return code;
}

static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
  (void)taosThreadMutexLock(&pDispatcher->mutex);
  pDispatcher->queryEnd = true;
  pDispatcher->useconds = useconds;
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
}

static void resetDispatcher(struct SDataSinkHandle* pHandle) {
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
  (void)taosThreadMutexLock(&pDispatcher->mutex);
  pDispatcher->queryEnd = false;
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);
}

static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRowLen, bool* pQueryEnd) {
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
  if (taosQueueEmpty(pDispatcher->pDataBlocks)) {
    *pQueryEnd = pDispatcher->queryEnd;
    *pLen = 0;
    return;
  }

  SDataDispatchBuf* pBuf = NULL;
  taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);
  if (pBuf != NULL) {
    TAOS_MEMCPY(&pDispatcher->nextOutput, pBuf, sizeof(SDataDispatchBuf));
    taosFreeQitem(pBuf);
  }

  SDataCacheEntry* pEntry = (SDataCacheEntry*)pDispatcher->nextOutput.pData;
  *pLen = pEntry->dataLen;
  *pRowLen = pEntry->rawLen;

  *pQueryEnd = pDispatcher->queryEnd;
  qDebug("got data len %" PRId64 ", row num %d in sink", *pLen,
         ((SDataCacheEntry*)(pDispatcher->nextOutput.pData))->numOfRows);
}

static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
  if (NULL == pDispatcher->nextOutput.pData) {
    if (!pDispatcher->queryEnd) {
      qError("empty res while query not end in data dispatcher");
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
    }

    pOutput->useconds = pDispatcher->useconds;
    pOutput->precision = pDispatcher->pSchema->precision;
    pOutput->bufStatus = DS_BUF_EMPTY;
    pOutput->queryEnd = pDispatcher->queryEnd;
    return TSDB_CODE_SUCCESS;
  }

  SDataCacheEntry* pEntry = (SDataCacheEntry*)(pDispatcher->nextOutput.pData);
  TAOS_MEMCPY(pOutput->pData, pEntry->data, pEntry->dataLen);
  pOutput->numOfRows = pEntry->numOfRows;
  pOutput->numOfCols = pEntry->numOfCols;
  pOutput->compressed = pEntry->compressed;

  (void)atomic_sub_fetch_64(&pDispatcher->cachedSize, pEntry->dataLen);
  (void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);

  taosMemoryFreeClear(pDispatcher->nextOutput.pData);  // todo persistent
  pOutput->bufStatus = updateStatus(pDispatcher);
  
  (void)taosThreadMutexLock(&pDispatcher->mutex);
  pOutput->queryEnd = pDispatcher->queryEnd;
  pOutput->useconds = pDispatcher->useconds;
  pOutput->precision = pDispatcher->pSchema->precision;
  (void)taosThreadMutexUnlock(&pDispatcher->mutex);

  return TSDB_CODE_SUCCESS;
}

static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
  (void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pDispatcher->cachedSize);
  taosMemoryFreeClear(pDispatcher->nextOutput.pData);
  nodesDestroyNode((SNode*)pDispatcher->pSinkNode);

  while (!taosQueueEmpty(pDispatcher->pDataBlocks)) {
    SDataDispatchBuf* pBuf = NULL;
    taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);
    if (pBuf != NULL) {
      taosMemoryFreeClear(pBuf->pData);
      taosFreeQitem(pBuf);
    }
  }

  taosCloseQueue(pDispatcher->pDataBlocks);
  taosMemoryFreeClear(pDispatcher->pCompressBuf);
  pDispatcher->bufSize = 0;

  (void)taosThreadMutexDestroy(&pDispatcher->mutex);
  taosMemoryFree(pDispatcher->pManager);
  return TSDB_CODE_SUCCESS;
}

static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;

  *size = atomic_load_64(&pDispatcher->cachedSize);
  return TSDB_CODE_SUCCESS;
}

static int32_t getSinkFlags(struct SDataSinkHandle* pHandle, uint64_t* pFlags) {
  SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;

  *pFlags = atomic_load_64(&pDispatcher->flags);
  return TSDB_CODE_SUCCESS;
}

static int32_t blockDescNodeCheck(SDataBlockDescNode* pInputDataBlockDesc)  {
  if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {
    return TSDB_CODE_SUCCESS;
  }

  if (pInputDataBlockDesc == NULL) {
    qError("invalid schema");
    return TSDB_CODE_QRY_INVALID_INPUT;
  }

  SNode*  pNode;
  int32_t realOutputRowSize = 0;
  FOREACH(pNode, pInputDataBlockDesc->pSlots) {
    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
    if (pSlotDesc->output) {
      realOutputRowSize += pSlotDesc->dataType.bytes;
    } else {
      // Slots must be sorted, and slots with 'output' set to true must come first
      break;
    }
  }
  if (realOutputRowSize !=  pInputDataBlockDesc->outputRowSize) {
    qError("invalid schema, realOutputRowSize:%d, outputRowSize:%d", realOutputRowSize, pInputDataBlockDesc->outputRowSize);
    return TSDB_CODE_QRY_INVALID_INPUT;
  }
  return TSDB_CODE_SUCCESS;
}

int32_t getOutputColCounts(SDataBlockDescNode* pInputDataBlockDesc) {
  if (pInputDataBlockDesc == NULL) {
    qError("invalid schema");
    return 0;
  }
  SNode*  pNode;
  int32_t numOfCols = 0;
  FOREACH(pNode, pInputDataBlockDesc->pSlots) {
    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;
    if (pSlotDesc->output) {
      ++numOfCols;
    } else {
      // Slots must be sorted, and slots with 'output' set to true must come first
      break;
    }
  }
  return numOfCols;
}

int32_t createDataDispatcher(SDataSinkManager* pManager, SDataSinkNode** ppDataSink, DataSinkHandle* pHandle, bool processOneBlock) {
  int32_t code;
  SDataSinkNode* pDataSink = *ppDataSink;
  code = blockDescNodeCheck(pDataSink->pInputDataBlockDesc);
  if (code) {
    qError("failed to check input data block desc, code:%d", code);
    goto _return;
  }

  SDataDispatcherNode* pDispatherNode = (SDataDispatcherNode*)pDataSink;
  SDataDispatchHandle* dispatcher = taosMemoryCalloc(1, sizeof(SDataDispatchHandle));
  if (NULL == dispatcher) {
    goto _return;
  }

  dispatcher->sink.fPut = putDataBlock;
  dispatcher->sink.fEndPut = endPut;
  dispatcher->sink.fReset = resetDispatcher;
  dispatcher->sink.fGetLen = getDataLength;
  dispatcher->sink.fGetData = getDataBlock;
  dispatcher->sink.fDestroy = destroyDataSinker;
  dispatcher->sink.fGetCacheSize = getCacheSize;
  dispatcher->sink.fGetFlags = getSinkFlags;
  dispatcher->pManager = pManager;
  pManager = NULL;
  dispatcher->dynamicSchema = pDispatherNode->dynamicSchema;
  dispatcher->pSchema = pDataSink->pInputDataBlockDesc;
  dispatcher->pSinkNode = pDataSink;
  *ppDataSink = NULL;
  dispatcher->outPutColCounts = getOutputColCounts(dispatcher->pSchema);
  dispatcher->status = DS_BUF_EMPTY;
  dispatcher->queryEnd = false;
  code = taosOpenQueue(&dispatcher->pDataBlocks);
  if (code) {
    terrno = code;
    goto _return;
  }
  code = taosThreadMutexInit(&dispatcher->mutex, NULL);
  if (code) {
    terrno = code;
    goto _return;
  }

  dispatcher->flags = DS_FLAG_USE_MEMPOOL;
  if (processOneBlock) {
    dispatcher->flags |= DS_FLAG_PROCESS_ONE_BLOCK;
  }

  *pHandle = dispatcher;
  return TSDB_CODE_SUCCESS;

_return:

  taosMemoryFree(pManager);
  
  if (dispatcher) {
    dsDestroyDataSinker(dispatcher);
  }

  nodesDestroyNode((SNode *)*ppDataSink);
  *ppDataSink = NULL;
  
  return terrno;
}

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) {	×
62	if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {	×
63	return TSDB_CODE_SUCCESS;	×
64	}
65	if (pInput == NULL \|\| pInput->pData == NULL \|\| pInput->pData->info.rows <= 0) {	×
66	qError("invalid input data");	×
67	return TSDB_CODE_QRY_INVALID_INPUT;	×
68	}
69	SDataBlockDescNode* pSchema = pHandle->pSchema;	×
70	if (pSchema == NULL \|\| pSchema->totalRowSize != pInput->pData->info.rowSize) {	×
71	qError("invalid schema");	×
72	return TSDB_CODE_QRY_INVALID_INPUT;	×
73	}
74
75	if (pHandle->outPutColCounts > taosArrayGetSize(pInput->pData->pDataBlock)) {	×
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;	×
82	FOREACH(pNode, pHandle->pSchema->pSlots) {	×
83	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	×
84	if (pSlotDesc->output) {	×
85	SColumnInfoData* pColInfoData = taosArrayGet(pInput->pData->pDataBlock, colNum);	×
86	if (pColInfoData == NULL) {	×
87	return -1;	×
88	}
89	if (pColInfoData->info.bytes < 0) {	×
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) &&	×
94	TYPE_BYTES[pColInfoData->info.type] != pColInfoData->info.bytes) {	×
95	qError("invalid column bytes, schema:%d, input:%d", TYPE_BYTES[pColInfoData->info.type],	×
96	pColInfoData->info.bytes);
97	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
98	}
99	if (pColInfoData->info.type != pSlotDesc->dataType.type) {	×
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) {	×
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)) {	×
109	qError("invalid column type, type:%d", pColInfoData->info.type);	×
110	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
111	}
112	++colNum;	×
113	}
114	}
115
116
117	return TSDB_CODE_SUCCESS;	×
118	}
119
120	static int32_t inputSafetyCheckForDynamicSchema(const SInputData* pInput) {	×
121	if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {	×
122	return TSDB_CODE_SUCCESS;	×
123	}
124	if (pInput == NULL \|\| pInput->pData == NULL \|\| pInput->pData->info.rows <= 0) {	×
125	qError("invalid input data");	×
126	return TSDB_CODE_QRY_INVALID_INPUT;	×
127	}
128
129	for (int32_t colNum = 0; colNum < taosArrayGetSize(pInput->pData->pDataBlock); colNum++) {	×
130	SColumnInfoData* pColInfoData = taosArrayGet(pInput->pData->pDataBlock, colNum);	×
131	if (pColInfoData == NULL) {	×
132	return -1;	×
133	}
134	if (pColInfoData->info.bytes < 0) {	×
135	qError("invalid column bytes, input:%d", pColInfoData->info.bytes);	×
136	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
137	}
138	if (!IS_VAR_DATA_TYPE(pColInfoData->info.type) &&	×
139	TYPE_BYTES[pColInfoData->info.type] != pColInfoData->info.bytes) {	×
140	qError("invalid column bytes, schema:%d, input:%d", TYPE_BYTES[pColInfoData->info.type],	×
141	pColInfoData->info.bytes);
142	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
143	}
144
145	if (IS_INVALID_TYPE(pColInfoData->info.type)) {	×
146	qError("invalid column type, type:%d", pColInfoData->info.type);	×
147	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
148	}
149	}
150
151	return TSDB_CODE_SUCCESS;	×
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) {	×
164	int32_t numOfCols = 0;	×
165	SNode* pNode;
166	int32_t code = TSDB_CODE_SUCCESS;	×
167
168	if (pHandle->dynamicSchema) {	×
169	code = inputSafetyCheckForDynamicSchema(pInput);	×
170	if (code) {	×
171	qError("failed to check input data for dynamic schema, code:%d", code);	×
172	return code;	×
173	}
174	numOfCols = taosArrayGetSize(pInput->pData->pDataBlock);	×
175	} else {
176	code = inputSafetyCheck(pHandle, pInput);	×
177	if (code) {	×
178	qError("failed to check input data, code:%d", code);	×
179	return code;	×
180	}
181
182	FOREACH(pNode, pHandle->pSchema->pSlots) {	×
183	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	×
184	if (pSlotDesc->output) {	×
185	++numOfCols;	×
186	}
187	}
188	}
189
190	SDataCacheEntry* pEntry = (SDataCacheEntry*)pBuf->pData;	×
191	pEntry->compressed = 0;	×
192	pEntry->numOfRows = pInput->pData->info.rows;	×
193	pEntry->numOfCols = numOfCols;	×
194	pEntry->dataLen = 0;	×
195	pEntry->rawLen = 0;	×
196
197	pBuf->useSize = sizeof(SDataCacheEntry);	×
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;	×
202	if ((pBuf->allocSize > tsCompressMsgSize) && (tsCompressMsgSize > 0) && pHandle->pManager->cfg.compress) {	×
203	if (pHandle->pCompressBuf == NULL) {	×
204	pHandle->pCompressBuf = taosMemoryMalloc(dataEncodeBufSize);	×
205	if (NULL == pHandle->pCompressBuf) {	×
206	QRY_RET(terrno);	×
207	}
208	pHandle->bufSize = dataEncodeBufSize;	×
209	} else {
210	if (pHandle->bufSize < dataEncodeBufSize) {	×
211	pHandle->bufSize = dataEncodeBufSize;	×
212	void* p = taosMemoryRealloc(pHandle->pCompressBuf, pHandle->bufSize);	×
213	if (p != NULL) {	×
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
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);	×
225	return terrno;	×
226	}
227	int32_t len =
228	tsCompressString(pHandle->pCompressBuf, dataLen, 1, pEntry->data, pBuf->allocSize, ONE_STAGE_COMP, NULL, 0);	×
229	if (len < dataLen) {	×
230	pEntry->compressed = 1;	×
231	pEntry->dataLen = len;	×
232	pEntry->rawLen = dataLen;	×
233	} else { // no need to compress data
234	pEntry->compressed = 0;	×
235	pEntry->dataLen = dataLen;	×
236	pEntry->rawLen = dataLen;	×
237	TAOS_MEMCPY(pEntry->data, pHandle->pCompressBuf, dataLen);	×
238	}
239	} else {
240	pEntry->dataLen =	×
241	pHandle->dynamicSchema	×
242	? blockEncodeInternal(pInput->pData, pEntry->data, pBuf->allocSize, numOfCols)	×
243	: blockEncode(pInput->pData, pEntry->data, pBuf->allocSize, numOfCols);	×
244	if(pEntry->dataLen < 0) {	×
245	qError("failed to encode data block, code: %d", pEntry->dataLen);	×
246	return terrno;	×
247	}
248	pEntry->rawLen = pEntry->dataLen;	×
249	}
250	}
251
252	pBuf->useSize += pEntry->dataLen;	×
253
254	(void)atomic_add_fetch_64(&pHandle->cachedSize, pEntry->dataLen);	×
255	(void)atomic_add_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);	×
256
257	return TSDB_CODE_SUCCESS;	×
258	}
259
260	static int32_t allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput, SDataDispatchBuf* pBuf) {	×
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	×
271	? blockGetInternalEncodeSize(pInput->pData)	×
272	: blockGetEncodeSize(pInput->pData));	×
273
274	pBuf->pData = taosMemoryMalloc(pBuf->allocSize);	×
275	if (pBuf->pData == NULL) {	×
276	qError("SinkNode failed to malloc memory, size:%d, code:%x", pBuf->allocSize, terrno);	×
277	return terrno;	×
278	}
279
280	return TSDB_CODE_SUCCESS;	×
281	}
282
283	static int32_t updateStatus(SDataDispatchHandle* pDispatcher) {	×
284	(void)taosThreadMutexLock(&pDispatcher->mutex);	×
285	int32_t blockNums = taosQueueItemSize(pDispatcher->pDataBlocks);	×
286	int32_t status =	×
287	(0 == blockNums ? DS_BUF_EMPTY
288	: (blockNums < pDispatcher->pManager->cfg.maxDataBlockNumPerQuery ? DS_BUF_LOW : DS_BUF_FULL));	×
289	pDispatcher->status = status;	×
290	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	×
291	return status;	×
292	}
293
294	static int32_t getStatus(SDataDispatchHandle* pDispatcher) {	×
295	(void)taosThreadMutexLock(&pDispatcher->mutex);	×
296	int32_t status = pDispatcher->status;	×
297	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	×
298	return status;	×
299	}
300
301	static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {	×
302	int32_t code = 0;	×
303	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
304	SDataDispatchBuf* pBuf = NULL;	×
305
306	code = taosAllocateQitem(sizeof(SDataDispatchBuf), DEF_QITEM, 0, (void**)&pBuf);	×
307	if (code) {	×
308	return code;	×
309	}
310
311	code = allocBuf(pDispatcher, pInput, pBuf);	×
312	if (code) {	×
313	taosFreeQitem(pBuf);	×
314	return code;	×
315	}
316
317	QRY_ERR_JRET(toDataCacheEntry(pDispatcher, pInput, pBuf));	×
318	QRY_ERR_JRET(taosWriteQitem(pDispatcher->pDataBlocks, pBuf));	×
319
320	int32_t status = updateStatus(pDispatcher);	×
321	*pContinue = (status == DS_BUF_LOW \|\| status == DS_BUF_EMPTY) && !(pDispatcher->flags & DS_FLAG_PROCESS_ONE_BLOCK);	×
322	return TSDB_CODE_SUCCESS;	×
323
324	_return:	×
325
326	taosMemoryFreeClear(pBuf->pData);	×
327	taosFreeQitem(pBuf);	×
328	return code;	×
329	}
330
331	static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {	×
332	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
333	(void)taosThreadMutexLock(&pDispatcher->mutex);	×
334	pDispatcher->queryEnd = true;	×
335	pDispatcher->useconds = useconds;	×
336	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	×
337	}	×
338
339	static void resetDispatcher(struct SDataSinkHandle* pHandle) {	×
340	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
341	(void)taosThreadMutexLock(&pDispatcher->mutex);	×
342	pDispatcher->queryEnd = false;	×
343	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	×
344	}	×
345
346	static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRowLen, bool* pQueryEnd) {	×
347	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
348	if (taosQueueEmpty(pDispatcher->pDataBlocks)) {	×
349	*pQueryEnd = pDispatcher->queryEnd;	×
350	*pLen = 0;	×
351	return;	×
352	}
353
354	SDataDispatchBuf* pBuf = NULL;	×
355	taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);	×
356	if (pBuf != NULL) {	×
357	TAOS_MEMCPY(&pDispatcher->nextOutput, pBuf, sizeof(SDataDispatchBuf));	×
358	taosFreeQitem(pBuf);	×
359	}
360
361	SDataCacheEntry* pEntry = (SDataCacheEntry*)pDispatcher->nextOutput.pData;	×
362	*pLen = pEntry->dataLen;	×
363	*pRowLen = pEntry->rawLen;	×
364
365	*pQueryEnd = pDispatcher->queryEnd;	×
366	qDebug("got data len %" PRId64 ", row num %d in sink", *pLen,	×
367	((SDataCacheEntry*)(pDispatcher->nextOutput.pData))->numOfRows);
368	}
369
370	static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {	×
371	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
372	if (NULL == pDispatcher->nextOutput.pData) {	×
373	if (!pDispatcher->queryEnd) {	×
374	qError("empty res while query not end in data dispatcher");	×
375	return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
376	}
377
378	pOutput->useconds = pDispatcher->useconds;	×
379	pOutput->precision = pDispatcher->pSchema->precision;	×
380	pOutput->bufStatus = DS_BUF_EMPTY;	×
381	pOutput->queryEnd = pDispatcher->queryEnd;	×
382	return TSDB_CODE_SUCCESS;	×
383	}
384
385	SDataCacheEntry* pEntry = (SDataCacheEntry*)(pDispatcher->nextOutput.pData);	×
386	TAOS_MEMCPY(pOutput->pData, pEntry->data, pEntry->dataLen);	×
387	pOutput->numOfRows = pEntry->numOfRows;	×
388	pOutput->numOfCols = pEntry->numOfCols;	×
389	pOutput->compressed = pEntry->compressed;	×
390
391	(void)atomic_sub_fetch_64(&pDispatcher->cachedSize, pEntry->dataLen);	×
392	(void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);	×
393
394	taosMemoryFreeClear(pDispatcher->nextOutput.pData); // todo persistent	×
395	pOutput->bufStatus = updateStatus(pDispatcher);	×
396
397	(void)taosThreadMutexLock(&pDispatcher->mutex);	×
398	pOutput->queryEnd = pDispatcher->queryEnd;	×
399	pOutput->useconds = pDispatcher->useconds;	×
400	pOutput->precision = pDispatcher->pSchema->precision;	×
401	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	×
402
403	return TSDB_CODE_SUCCESS;	×
404	}
405
406	static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {	×
407	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
408	(void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pDispatcher->cachedSize);	×
409	taosMemoryFreeClear(pDispatcher->nextOutput.pData);	×
410	nodesDestroyNode((SNode*)pDispatcher->pSinkNode);	×
411
412	while (!taosQueueEmpty(pDispatcher->pDataBlocks)) {	×
413	SDataDispatchBuf* pBuf = NULL;	×
414	taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);	×
415	if (pBuf != NULL) {	×
416	taosMemoryFreeClear(pBuf->pData);	×
417	taosFreeQitem(pBuf);	×
418	}
419	}
420
421	taosCloseQueue(pDispatcher->pDataBlocks);	×
422	taosMemoryFreeClear(pDispatcher->pCompressBuf);	×
423	pDispatcher->bufSize = 0;	×
424
425	(void)taosThreadMutexDestroy(&pDispatcher->mutex);	×
426	taosMemoryFree(pDispatcher->pManager);	×
427	return TSDB_CODE_SUCCESS;	×
428	}
429
430	static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {	×
431	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
432
433	*size = atomic_load_64(&pDispatcher->cachedSize);	×
434	return TSDB_CODE_SUCCESS;	×
435	}
436
437	static int32_t getSinkFlags(struct SDataSinkHandle* pHandle, uint64_t* pFlags) {	×
438	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
439
440	*pFlags = atomic_load_64(&pDispatcher->flags);	×
441	return TSDB_CODE_SUCCESS;	×
442	}
443
444	static int32_t blockDescNodeCheck(SDataBlockDescNode* pInputDataBlockDesc) {	×
445	if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {	×
446	return TSDB_CODE_SUCCESS;	×
447	}
448
449	if (pInputDataBlockDesc == NULL) {	×
450	qError("invalid schema");	×
451	return TSDB_CODE_QRY_INVALID_INPUT;	×
452	}
453
454	SNode* pNode;
455	int32_t realOutputRowSize = 0;	×
456	FOREACH(pNode, pInputDataBlockDesc->pSlots) {	×
457	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	×
458	if (pSlotDesc->output) {	×
459	realOutputRowSize += pSlotDesc->dataType.bytes;	×
460	} else {
461	// Slots must be sorted, and slots with 'output' set to true must come first
462	break;	×
463	}
464	}
465	if (realOutputRowSize != pInputDataBlockDesc->outputRowSize) {	×
466	qError("invalid schema, realOutputRowSize:%d, outputRowSize:%d", realOutputRowSize, pInputDataBlockDesc->outputRowSize);	×
467	return TSDB_CODE_QRY_INVALID_INPUT;	×
468	}
469	return TSDB_CODE_SUCCESS;	×
470	}
471
472	int32_t getOutputColCounts(SDataBlockDescNode* pInputDataBlockDesc) {	×
473	if (pInputDataBlockDesc == NULL) {	×
474	qError("invalid schema");	×
475	return 0;	×
476	}
477	SNode* pNode;
478	int32_t numOfCols = 0;	×
479	FOREACH(pNode, pInputDataBlockDesc->pSlots) {	×
480	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	×
481	if (pSlotDesc->output) {	×
482	++numOfCols;	×
483	} else {
484	// Slots must be sorted, and slots with 'output' set to true must come first
485	break;	×
486	}
487	}
488	return numOfCols;	×
489	}
490
491	int32_t createDataDispatcher(SDataSinkManager* pManager, SDataSinkNode** ppDataSink, DataSinkHandle* pHandle, bool processOneBlock) {	×
492	int32_t code;
493	SDataSinkNode* pDataSink = *ppDataSink;	×
494	code = blockDescNodeCheck(pDataSink->pInputDataBlockDesc);	×
495	if (code) {	×
496	qError("failed to check input data block desc, code:%d", code);	×
497	goto _return;	×
498	}
499
500	SDataDispatcherNode* pDispatherNode = (SDataDispatcherNode*)pDataSink;	×
501	SDataDispatchHandle* dispatcher = taosMemoryCalloc(1, sizeof(SDataDispatchHandle));	×
502	if (NULL == dispatcher) {	×
503	goto _return;	×
504	}
505
506	dispatcher->sink.fPut = putDataBlock;	×
507	dispatcher->sink.fEndPut = endPut;	×
508	dispatcher->sink.fReset = resetDispatcher;	×
509	dispatcher->sink.fGetLen = getDataLength;	×
510	dispatcher->sink.fGetData = getDataBlock;	×
511	dispatcher->sink.fDestroy = destroyDataSinker;	×
512	dispatcher->sink.fGetCacheSize = getCacheSize;	×
513	dispatcher->sink.fGetFlags = getSinkFlags;	×
514	dispatcher->pManager = pManager;	×
515	pManager = NULL;	×
516	dispatcher->dynamicSchema = pDispatherNode->dynamicSchema;	×
517	dispatcher->pSchema = pDataSink->pInputDataBlockDesc;	×
518	dispatcher->pSinkNode = pDataSink;	×
519	*ppDataSink = NULL;	×
520	dispatcher->outPutColCounts = getOutputColCounts(dispatcher->pSchema);	×
521	dispatcher->status = DS_BUF_EMPTY;	×
522	dispatcher->queryEnd = false;	×
523	code = taosOpenQueue(&dispatcher->pDataBlocks);	×
524	if (code) {	×
525	terrno = code;	×
526	goto _return;	×
527	}
528	code = taosThreadMutexInit(&dispatcher->mutex, NULL);	×
529	if (code) {	×
530	terrno = code;	×
531	goto _return;	×
532	}
533
534	dispatcher->flags = DS_FLAG_USE_MEMPOOL;	×
535	if (processOneBlock) {	×
536	dispatcher->flags \|= DS_FLAG_PROCESS_ONE_BLOCK;	×
537	}
538
539	*pHandle = dispatcher;	×
540	return TSDB_CODE_SUCCESS;	×
541
542	_return:	×
543
544	taosMemoryFree(pManager);	×
545
546	if (dispatcher) {	×
547	dsDestroyDataSinker(dispatcher);	×
548	}
549
550	nodesDestroyNode((SNode )ppDataSink);	×
551	*ppDataSink = NULL;	×
552
553	return terrno;	×
554	}

taosdata / TDengine / #4986

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

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