#3522

Committed 07 Nov 2024 05:59AM UTC coverage: 58.216% (+1.3%) from 56.943%

Build # #3522

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28663 from taosdata/fix/3_liaohj

fix(stream): stop the underlying scan operations for stream

Run Details

111884 of 248391 branches covered (45.04%)

Branch coverage included in aggregate %.

3 of 4 new or added lines in 1 file covered. (75.0%)

1164 existing lines in 134 files now uncovered.

191720 of 273118 relevant lines covered (70.2%)

13088725.13 hits per line

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

58.18

/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;
  STaosQueue*         pDataBlocks;
  SDataDispatchBuf    nextOutput;
  int32_t             outPutColCounts;
  int32_t             status;
  bool                queryEnd;
  uint64_t            useconds;
  uint64_t            cachedSize;
  void*               pCompressBuf;
  int32_t             bufSize;
  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;
}

// 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 = 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 = 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) + 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);
  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);

  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 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, const SDataSinkNode* pDataSink, DataSinkHandle* pHandle) {
  int32_t code;
  code = blockDescNodeCheck(pDataSink->pInputDataBlockDesc);
  if (code) {
    qError("failed to check input data block desc, code:%d", code);
    return code;
  }

  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->pManager = pManager;
  pManager = NULL;
  dispatcher->pSchema = pDataSink->pInputDataBlockDesc;
  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;
  }

  *pHandle = dispatcher;
  return TSDB_CODE_SUCCESS;

_return:

  taosMemoryFree(pManager);
  
  if (dispatcher) {
    dsDestroyDataSinker(dispatcher);
  }
  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	STaosQueue* pDataBlocks;
47	SDataDispatchBuf nextOutput;
48	int32_t outPutColCounts;
49	int32_t status;
50	bool queryEnd;
51	uint64_t useconds;
52	uint64_t cachedSize;
53	void* pCompressBuf;
54	int32_t bufSize;
55	TdThreadMutex mutex;
56	} SDataDispatchHandle;
57
58	static int32_t inputSafetyCheck(SDataDispatchHandle* pHandle, const SInputData* pInput) {	11,142,130✔
59	if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {	11,142,130!
60	return TSDB_CODE_SUCCESS;	×
61	}
62	if (pInput == NULL \|\| pInput->pData == NULL \|\| pInput->pData->info.rows <= 0) {	11,142,130!
63	qError("invalid input data");	1!
64	return TSDB_CODE_QRY_INVALID_INPUT;	×
65	}
66	SDataBlockDescNode* pSchema = pHandle->pSchema;	11,142,129✔
67	if (pSchema == NULL \|\| pSchema->totalRowSize != pInput->pData->info.rowSize) {	11,142,129!
68	qError("invalid schema");	19!
69	return TSDB_CODE_QRY_INVALID_INPUT;	×
70	}
71
72	if (pHandle->outPutColCounts > taosArrayGetSize(pInput->pData->pDataBlock)) {	11,142,110✔
73	qError("invalid column number, schema:%d, input:%zu", pHandle->outPutColCounts, taosArrayGetSize(pInput->pData->pDataBlock));	30!
74	return TSDB_CODE_QRY_INVALID_INPUT;	×
75	}
76
77	SNode* pNode;
78	int32_t colNum = 0;	11,141,684✔
79	FOREACH(pNode, pHandle->pSchema->pSlots) {	47,263,888!
80	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	36,124,296✔
81	if (pSlotDesc->output) {	36,124,296✔
82	SColumnInfoData* pColInfoData = taosArrayGet(pInput->pData->pDataBlock, colNum);	35,593,561✔
83	if (pColInfoData == NULL) {	35,591,251!
84	return -1;	×
85	}
86	if (pColInfoData->info.bytes < 0) {	35,591,251!
87	qError("invalid column bytes, schema:%d, input:%d", pSlotDesc->dataType.bytes, pColInfoData->info.bytes);	×
88	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
89	}
90	if (!IS_VAR_DATA_TYPE(pColInfoData->info.type) &&	35,591,251!
91	TYPE_BYTES[pColInfoData->info.type] != pColInfoData->info.bytes) {	27,360,436!
92	qError("invalid column bytes, schema:%d, input:%d", TYPE_BYTES[pColInfoData->info.type],	×
93	pColInfoData->info.bytes);
94	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
95	}
96	if (pColInfoData->info.type != pSlotDesc->dataType.type) {	35,591,251!
97	qError("invalid column type, schema:%d, input:%d", pSlotDesc->dataType.type, pColInfoData->info.type);	×
98	return TSDB_CODE_QRY_INVALID_INPUT;	×
99	}
100	if (pColInfoData->info.bytes != pSlotDesc->dataType.bytes) {	35,591,251!
101	qError("invalid column bytes, schema:%d, input:%d", pSlotDesc->dataType.bytes, pColInfoData->info.bytes);	×
102	return TSDB_CODE_QRY_INVALID_INPUT;	×
103	}
104
105	if (IS_INVALID_TYPE(pColInfoData->info.type)) {	35,591,251!
106	qError("invalid column type, type:%d", pColInfoData->info.type);	×
107	return TSDB_CODE_TSC_INTERNAL_ERROR;	×
108	}
109	++colNum;	35,591,469✔
110	}
111	}
112
113
114	return TSDB_CODE_SUCCESS;	11,139,592✔
115	}
116
117	// clang-format off
118	// data format:
119	// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+
120	// \|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 \|
121	// \| \| sizeof(int32_t) \|sizeof(int32) \| sizeof(int32)\| sizeof(uint64_t) \| (sizeof(int8_t)+sizeof(int32_t))numOfCols \| sizeof(int32_t) numOfCols \| actual size \| \| \|
122	// +----------------+------------------+--------------+--------------+------------------+--------------------------------------------+------------------------------------+-------------+-----------+-------------+-----------+
123	// The length of bitmap is decided by number of rows of this data block, and the length of each column data is
124	// recorded in the first segment, next to the struct header
125	// clang-format on
126	static int32_t toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) {	11,142,283✔
127	int32_t numOfCols = 0;	11,142,283✔
128	SNode* pNode;
129
130	int32_t code = inputSafetyCheck(pHandle, pInput);	11,142,283✔
131	if (code) {	11,138,697!
UNCOV 132	qError("failed to check input data, code:%d", code);	×
133	return code;	×
134	}
135
136	FOREACH(pNode, pHandle->pSchema->pSlots) {	47,269,999!
137	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	36,131,166✔
138	if (pSlotDesc->output) {	36,131,166✔
139	++numOfCols;	35,600,031✔
140	}
141	}
142
143	SDataCacheEntry* pEntry = (SDataCacheEntry*)pBuf->pData;	11,138,833✔
144	pEntry->compressed = 0;	11,138,833✔
145	pEntry->numOfRows = pInput->pData->info.rows;	11,138,833✔
146	pEntry->numOfCols = numOfCols;	11,138,833✔
147	pEntry->dataLen = 0;	11,138,833✔
148	pEntry->rawLen = 0;	11,138,833✔
149
150	pBuf->useSize = sizeof(SDataCacheEntry);	11,138,833✔
151
152	{
153	// allocate additional 8 bytes to avoid invalid write if compress failed to reduce the size
154	size_t dataEncodeBufSize = pBuf->allocSize + 8;	11,138,833✔
155	if ((pBuf->allocSize > tsCompressMsgSize) && (tsCompressMsgSize > 0) && pHandle->pManager->cfg.compress) {	11,138,833!
156	if (pHandle->pCompressBuf == NULL) {	×
157	pHandle->pCompressBuf = taosMemoryMalloc(dataEncodeBufSize);	×
158	if (NULL == pHandle->pCompressBuf) {	×
159	QRY_RET(terrno);	×
160	}
161	pHandle->bufSize = dataEncodeBufSize;	×
162	} else {
163	if (pHandle->bufSize < dataEncodeBufSize) {	×
164	pHandle->bufSize = dataEncodeBufSize;	×
165	void* p = taosMemoryRealloc(pHandle->pCompressBuf, pHandle->bufSize);	×
166	if (p != NULL) {	×
167	pHandle->pCompressBuf = p;	×
168	} else {
169	qError("failed to prepare compress buf:%d, code: %x", pHandle->bufSize, terrno);	×
170	return terrno;	×
171	}
172	}
173	}
174
175	int32_t dataLen = blockEncode(pInput->pData, pHandle->pCompressBuf, dataEncodeBufSize, numOfCols);	×
176	if(dataLen < 0) {	×
177	qError("failed to encode data block, code: %d", dataLen);	×
178	return terrno;	×
179	}
180	int32_t len =
181	tsCompressString(pHandle->pCompressBuf, dataLen, 1, pEntry->data, pBuf->allocSize, ONE_STAGE_COMP, NULL, 0);	×
182	if (len < dataLen) {	×
183	pEntry->compressed = 1;	×
184	pEntry->dataLen = len;	×
185	pEntry->rawLen = dataLen;	×
186	} else { // no need to compress data
187	pEntry->compressed = 0;	×
188	pEntry->dataLen = dataLen;	×
189	pEntry->rawLen = dataLen;	×
190	TAOS_MEMCPY(pEntry->data, pHandle->pCompressBuf, dataLen);	×
191	}
192	} else {
193	pEntry->dataLen = blockEncode(pInput->pData, pEntry->data, pBuf->allocSize, numOfCols);	11,138,833✔
194	if(pEntry->dataLen < 0) {	11,139,678✔
195	qError("failed to encode data block, code: %d", pEntry->dataLen);	99!
196	return terrno;	99✔
197	}
198	pEntry->rawLen = pEntry->dataLen;	11,139,579✔
199	}
200	}
201
202	pBuf->useSize += pEntry->dataLen;	11,139,579✔
203
204	(void)atomic_add_fetch_64(&pHandle->cachedSize, pEntry->dataLen);	11,139,579✔
205	(void)atomic_add_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);	11,143,282✔
206
207	return TSDB_CODE_SUCCESS;	11,143,335✔
208	}
209
210	static int32_t allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput, SDataDispatchBuf* pBuf) {	11,143,684✔
211	/*
212	uint32_t capacity = pDispatcher->pManager->cfg.maxDataBlockNumPerQuery;
213	if (taosQueueItemSize(pDispatcher->pDataBlocks) > capacity) {
214	qError("SinkNode queue is full, no capacity, max:%d, current:%d, no capacity", capacity,
215	taosQueueItemSize(pDispatcher->pDataBlocks));
216	return false;
217	}
218	*/
219
220	pBuf->allocSize = sizeof(SDataCacheEntry) + blockGetEncodeSize(pInput->pData);	11,143,684✔
221
222	pBuf->pData = taosMemoryMalloc(pBuf->allocSize);	11,138,170✔
223	if (pBuf->pData == NULL) {	11,142,585!
224	qError("SinkNode failed to malloc memory, size:%d, code:%x", pBuf->allocSize, terrno);	×
225	return terrno;	×
226	}
227
228	return TSDB_CODE_SUCCESS;	11,142,585✔
229	}
230
231	static int32_t updateStatus(SDataDispatchHandle* pDispatcher) {	21,948,793✔
232	(void)taosThreadMutexLock(&pDispatcher->mutex);	21,948,793✔
233	int32_t blockNums = taosQueueItemSize(pDispatcher->pDataBlocks);	21,955,325✔
234	int32_t status =	21,964,075✔
235	(0 == blockNums ? DS_BUF_EMPTY
236	: (blockNums < pDispatcher->pManager->cfg.maxDataBlockNumPerQuery ? DS_BUF_LOW : DS_BUF_FULL));	21,964,075✔
237	pDispatcher->status = status;	21,964,075✔
238	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	21,964,075✔
239	return status;	21,968,441✔
240	}
241
242	static int32_t getStatus(SDataDispatchHandle* pDispatcher) {	×
243	(void)taosThreadMutexLock(&pDispatcher->mutex);	×
244	int32_t status = pDispatcher->status;	×
245	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	×
246	return status;	×
247	}
248
249	static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {	11,141,467✔
250	int32_t code = 0;	11,141,467✔
251	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	11,141,467✔
252	SDataDispatchBuf* pBuf = NULL;	11,141,467✔
253
254	code = taosAllocateQitem(sizeof(SDataDispatchBuf), DEF_QITEM, 0, (void**)&pBuf);	11,141,467✔
255	if (code) {	11,143,735!
256	return code;	×
257	}
258
259	code = allocBuf(pDispatcher, pInput, pBuf);	11,143,735✔
260	if (code) {	11,142,410!
261	taosFreeQitem(pBuf);	×
262	return code;	×
263	}
264
265	QRY_ERR_JRET(toDataCacheEntry(pDispatcher, pInput, pBuf));	11,142,410!
266	QRY_ERR_JRET(taosWriteQitem(pDispatcher->pDataBlocks, pBuf));	11,143,332!
267
268	int32_t status = updateStatus(pDispatcher);	11,143,778✔
269	*pContinue = (status == DS_BUF_LOW \|\| status == DS_BUF_EMPTY);	11,143,575✔
270	return TSDB_CODE_SUCCESS;	11,143,575✔
271
272	_return:	×
273
274	taosMemoryFreeClear(pBuf->pData);	×
275	taosFreeQitem(pBuf);	×
276	return code;	×
277	}
278
279	static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {	5,307,008✔
280	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	5,307,008✔
281	(void)taosThreadMutexLock(&pDispatcher->mutex);	5,307,008✔
282	pDispatcher->queryEnd = true;	5,307,555✔
283	pDispatcher->useconds = useconds;	5,307,555✔
284	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	5,307,555✔
285	}	5,307,564✔
286
287	static void resetDispatcher(struct SDataSinkHandle* pHandle) {	591✔
288	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	591✔
289	(void)taosThreadMutexLock(&pDispatcher->mutex);	591✔
290	pDispatcher->queryEnd = false;	592✔
291	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	592✔
292	}	592✔
293
294	static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRowLen, bool* pQueryEnd) {	12,639,941✔
295	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	12,639,941✔
296	if (taosQueueEmpty(pDispatcher->pDataBlocks)) {	12,639,941✔
297	*pQueryEnd = pDispatcher->queryEnd;	1,848,347✔
298	*pLen = 0;	1,848,347✔
299	return;	1,848,347✔
300	}
301
302	SDataDispatchBuf* pBuf = NULL;	10,810,949✔
303	taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);	10,810,949✔
304	if (pBuf != NULL) {	10,817,991✔
305	TAOS_MEMCPY(&pDispatcher->nextOutput, pBuf, sizeof(SDataDispatchBuf));	10,816,554✔
306	taosFreeQitem(pBuf);	10,816,554✔
307	}
308
309	SDataCacheEntry* pEntry = (SDataCacheEntry*)pDispatcher->nextOutput.pData;	10,836,890✔
310	*pLen = pEntry->dataLen;	10,836,890✔
311	*pRowLen = pEntry->rawLen;	10,836,890✔
312
313	*pQueryEnd = pDispatcher->queryEnd;	10,836,890✔
314	qDebug("got data len %" PRId64 ", row num %d in sink", *pLen,	10,836,890✔
315	((SDataCacheEntry*)(pDispatcher->nextOutput.pData))->numOfRows);
316	}
317
318	static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {	12,531,819✔
319	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	12,531,819✔
320	if (NULL == pDispatcher->nextOutput.pData) {	12,531,819✔
321	if (!pDispatcher->queryEnd) {	1,757,322!
322	qError("empty res while query not end in data dispatcher");	×
323	return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
324	}
325
326	pOutput->useconds = pDispatcher->useconds;	1,757,322✔
327	pOutput->precision = pDispatcher->pSchema->precision;	1,757,322✔
328	pOutput->bufStatus = DS_BUF_EMPTY;	1,757,322✔
329	pOutput->queryEnd = pDispatcher->queryEnd;	1,757,322✔
330	return TSDB_CODE_SUCCESS;	1,757,322✔
331	}
332
333	SDataCacheEntry* pEntry = (SDataCacheEntry*)(pDispatcher->nextOutput.pData);	10,774,497✔
334	TAOS_MEMCPY(pOutput->pData, pEntry->data, pEntry->dataLen);	10,774,497✔
335	pOutput->numOfRows = pEntry->numOfRows;	10,774,497✔
336	pOutput->numOfCols = pEntry->numOfCols;	10,774,497✔
337	pOutput->compressed = pEntry->compressed;	10,774,497✔
338
339	(void)atomic_sub_fetch_64(&pDispatcher->cachedSize, pEntry->dataLen);	10,774,497✔
340	(void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);	10,804,560✔
341
342	taosMemoryFreeClear(pDispatcher->nextOutput.pData); // todo persistent	10,829,531!
343	pOutput->bufStatus = updateStatus(pDispatcher);	10,821,447✔
344
345	(void)taosThreadMutexLock(&pDispatcher->mutex);	10,825,336✔
346	pOutput->queryEnd = pDispatcher->queryEnd;	10,823,031✔
347	pOutput->useconds = pDispatcher->useconds;	10,823,031✔
348	pOutput->precision = pDispatcher->pSchema->precision;	10,823,031✔
349	(void)taosThreadMutexUnlock(&pDispatcher->mutex);	10,823,031✔
350
351	return TSDB_CODE_SUCCESS;	10,820,183✔
352	}
353
354	static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {	5,310,740✔
355	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	5,310,740✔
356	(void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pDispatcher->cachedSize);	5,310,740✔
357	taosMemoryFreeClear(pDispatcher->nextOutput.pData);	5,310,935!
358
359	while (!taosQueueEmpty(pDispatcher->pDataBlocks)) {	5,600,315✔
360	SDataDispatchBuf* pBuf = NULL;	289,339✔
361	taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);	289,339✔
362	if (pBuf != NULL) {	289,339!
363	taosMemoryFreeClear(pBuf->pData);	289,339!
364	taosFreeQitem(pBuf);	289,339✔
365	}
366	}
367
368	taosCloseQueue(pDispatcher->pDataBlocks);	5,310,940✔
369	taosMemoryFreeClear(pDispatcher->pCompressBuf);	5,311,079!
370	pDispatcher->bufSize = 0;	5,311,079✔
371
372	(void)taosThreadMutexDestroy(&pDispatcher->mutex);	5,311,079✔
373	taosMemoryFree(pDispatcher->pManager);	5,310,916✔
374	return TSDB_CODE_SUCCESS;	5,311,147✔
375	}
376
377	static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {	×
378	SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;	×
379
380	*size = atomic_load_64(&pDispatcher->cachedSize);	×
381	return TSDB_CODE_SUCCESS;	×
382	}
383
384	static int32_t blockDescNodeCheck(SDataBlockDescNode* pInputDataBlockDesc) {	5,296,590✔
385	if(tsSafetyCheckLevel == TSDB_SAFETY_CHECK_LEVELL_NEVER) {	5,296,590!
386	return TSDB_CODE_SUCCESS;	×
387	}
388
389	if (pInputDataBlockDesc == NULL) {	5,296,590!
390	qError("invalid schema");	×
391	return TSDB_CODE_QRY_INVALID_INPUT;	×
392	}
393
394	SNode* pNode;
395	int32_t realOutputRowSize = 0;	5,296,590✔
396	FOREACH(pNode, pInputDataBlockDesc->pSlots) {	33,504,717!
397	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	28,504,775✔
398	if (pSlotDesc->output) {	28,504,775✔
399	realOutputRowSize += pSlotDesc->dataType.bytes;	28,208,127✔
400	} else {
401	// Slots must be sorted, and slots with 'output' set to true must come first
402	break;	296,648✔
403	}
404	}
405	if (realOutputRowSize != pInputDataBlockDesc->outputRowSize) {	5,296,590!
406	qError("invalid schema, realOutputRowSize:%d, outputRowSize:%d", realOutputRowSize, pInputDataBlockDesc->outputRowSize);	×
407	return TSDB_CODE_QRY_INVALID_INPUT;	×
408	}
409	return TSDB_CODE_SUCCESS;	5,296,590✔
410	}
411
412	int32_t getOutputColCounts(SDataBlockDescNode* pInputDataBlockDesc) {	5,302,843✔
413	if (pInputDataBlockDesc == NULL) {	5,302,843!
414	qError("invalid schema");	×
415	return 0;	×
416	}
417	SNode* pNode;
418	int32_t numOfCols = 0;	5,302,843✔
419	FOREACH(pNode, pInputDataBlockDesc->pSlots) {	33,519,466!
420	SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;	28,513,272✔
421	if (pSlotDesc->output) {	28,513,272✔
422	++numOfCols;	28,216,623✔
423	} else {
424	// Slots must be sorted, and slots with 'output' set to true must come first
425	break;	296,649✔
426	}
427	}
428	return numOfCols;	5,302,843✔
429	}
430
431	int32_t createDataDispatcher(SDataSinkManager* pManager, const SDataSinkNode* pDataSink, DataSinkHandle* pHandle) {	5,296,663✔
432	int32_t code;
433	code = blockDescNodeCheck(pDataSink->pInputDataBlockDesc);	5,296,663✔
434	if (code) {	5,309,017!
435	qError("failed to check input data block desc, code:%d", code);	×
436	return code;	×
437	}
438
439	SDataDispatchHandle* dispatcher = taosMemoryCalloc(1, sizeof(SDataDispatchHandle));	5,309,017✔
440	if (NULL == dispatcher) {	5,302,274!
441	goto _return;	×
442	}
443
444	dispatcher->sink.fPut = putDataBlock;	5,302,274✔
445	dispatcher->sink.fEndPut = endPut;	5,302,274✔
446	dispatcher->sink.fReset = resetDispatcher;	5,302,274✔
447	dispatcher->sink.fGetLen = getDataLength;	5,302,274✔
448	dispatcher->sink.fGetData = getDataBlock;	5,302,274✔
449	dispatcher->sink.fDestroy = destroyDataSinker;	5,302,274✔
450	dispatcher->sink.fGetCacheSize = getCacheSize;	5,302,274✔
451
452	dispatcher->pManager = pManager;	5,302,274✔
453	pManager = NULL;	5,302,274✔
454	dispatcher->pSchema = pDataSink->pInputDataBlockDesc;	5,302,274✔
455	dispatcher->outPutColCounts = getOutputColCounts(dispatcher->pSchema);	5,302,274✔
456	dispatcher->status = DS_BUF_EMPTY;	5,307,959✔
457	dispatcher->queryEnd = false;	5,307,959✔
458	code = taosOpenQueue(&dispatcher->pDataBlocks);	5,307,959✔
459	if (code) {	5,306,644!
460	terrno = code;	×
461	goto _return;	×
462	}
463	code = taosThreadMutexInit(&dispatcher->mutex, NULL);	5,306,644✔
464	if (code) {	5,302,984!
465	terrno = code;	×
466	goto _return;	×
467	}
468
469	*pHandle = dispatcher;	5,305,158✔
470	return TSDB_CODE_SUCCESS;	5,305,158✔
471
472	_return:	×
473
474	taosMemoryFree(pManager);	×
475
476	if (dispatcher) {	×
477	dsDestroyDataSinker(dispatcher);	×
478	}
479	return terrno;	×
480	}

taosdata / TDengine / #3522

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