From 9bee5e2cee0702753c1945779ef451d8d0735aec Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Wed, 8 Jun 2022 16:08:26 +0800 Subject: [PATCH] refactor: do some internal refactor. --- include/common/tdatablock.h | 4 +- source/client/src/clientImpl.c | 11 ++++++ source/common/src/tdatablock.c | 45 +++++++++++++++++------ source/dnode/mnode/impl/src/mndShow.c | 2 +- source/libs/executor/src/dataDispatcher.c | 8 ++-- source/libs/executor/src/executorimpl.c | 32 +--------------- source/libs/executor/src/tsort.c | 4 +- 7 files changed, 55 insertions(+), 51 deletions(-) diff --git a/include/common/tdatablock.h b/include/common/tdatablock.h index ab76cb404a..af46535c94 100644 --- a/include/common/tdatablock.h +++ b/include/common/tdatablock.h @@ -206,7 +206,7 @@ SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int3 size_t blockDataGetSize(const SSDataBlock* pBlock); size_t blockDataGetRowSize(SSDataBlock* pBlock); double blockDataGetSerialRowSize(const SSDataBlock* pBlock); -size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock); +size_t blockDataGetSerialMetaSize(uint32_t numOfCols); int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo); int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst); @@ -238,7 +238,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool const char* stbFullName, int32_t vgId); static FORCE_INLINE int32_t blockGetEncodeSize(const SSDataBlock* pBlock) { - return blockDataGetSerialMetaSize(pBlock) + blockDataGetSize(pBlock); + return blockDataGetSerialMetaSize(pBlock->info.numOfCols) + blockDataGetSize(pBlock); } static FORCE_INLINE int32_t blockCompressColData(SColumnInfoData* pColRes, int32_t numOfRows, char* data, diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 62026dd577..9da0c77653 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -1340,6 +1340,17 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32 uint64_t groupId = *(uint64_t*)p; p += sizeof(uint64_t); + // check fields + for(int32_t i = 0; i < numOfCols; ++i) { + int16_t type = *(int16_t*) p; + p += sizeof(int16_t); + + int32_t bytes = *(int32_t*) p; + p += sizeof(int32_t); + + ASSERT(type == pFields[i].type && bytes == pFields[i].bytes); + } + int32_t* colLength = (int32_t*)p; p += sizeof(int32_t) * numOfCols; diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 99ff4a4a42..8c51f1d50c 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -682,9 +682,9 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock) { * @param pBlock * @return */ -size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock) { - // | total rows/total length | block group id | each column length | - return sizeof(int32_t) + sizeof(uint64_t) + pBlock->info.numOfCols * sizeof(int32_t); +size_t blockDataGetSerialMetaSize(uint32_t numOfCols) { + // | total rows/total length | block group id | column schema | each column length | + return sizeof(int32_t) + sizeof(uint64_t) + numOfCols * (sizeof(int16_t) + sizeof(int32_t)) + numOfCols * sizeof(int32_t); } double blockDataGetSerialRowSize(const SSDataBlock* pBlock) { @@ -1244,7 +1244,7 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) { } size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize) { - int32_t payloadSize = pageSize - blockDataGetSerialMetaSize(pBlock); + int32_t payloadSize = pageSize - blockDataGetSerialMetaSize(pBlock->info.numOfCols); int32_t rowSize = pBlock->info.rowSize; @@ -1883,34 +1883,44 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols, int8_t needCompress) { + // todo extract method int32_t* actualLen = (int32_t*)data; data += sizeof(int32_t); uint64_t* groupId = (uint64_t*)data; data += sizeof(uint64_t); + for(int32_t i = 0; i < numOfCols; ++i) { + SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i); + *((int16_t*) data) = pColInfoData->info.type; + data += sizeof(int16_t); + + *((int32_t*) data) = pColInfoData->info.bytes; + data += sizeof(int32_t); + } + int32_t* colSizes = (int32_t*)data; data += numOfCols * sizeof(int32_t); - *dataLen = (numOfCols * sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t)); + *dataLen = blockDataGetSerialMetaSize(numOfCols);; int32_t numOfRows = pBlock->info.rows; for (int32_t col = 0; col < numOfCols; ++col) { SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, col); // copy the null bitmap + size_t metaSize = 0; if (IS_VAR_DATA_TYPE(pColRes->info.type)) { - size_t metaSize = numOfRows * sizeof(int32_t); + metaSize = numOfRows * sizeof(int32_t); memcpy(data, pColRes->varmeta.offset, metaSize); - data += metaSize; - (*dataLen) += metaSize; } else { - int32_t len = BitmapLen(numOfRows); - memcpy(data, pColRes->nullbitmap, len); - data += len; - (*dataLen) += len; + metaSize = BitmapLen(numOfRows); + memcpy(data, pColRes->nullbitmap, metaSize); } + data += metaSize; + (*dataLen) += metaSize; + if (needCompress) { colSizes[col] = blockCompressColData(pColRes, numOfRows, data, needCompress); data += colSizes[col]; @@ -1939,6 +1949,17 @@ const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t pBlock->info.groupId = *(uint64_t*)pStart; pStart += sizeof(uint64_t); + for(int32_t i = 0; i < numOfCols; ++i) { + SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i); + pColInfoData->info.type = *(int16_t*)pStart; + pStart += sizeof(int16_t); + + pColInfoData->info.bytes = *(int32_t*)pStart; + pStart += sizeof(int32_t); + } + + blockDataEnsureCapacity(pBlock, numOfRows); + int32_t* colLen = (int32_t*)pStart; pStart += sizeof(int32_t) * numOfCols; diff --git a/source/dnode/mnode/impl/src/mndShow.c b/source/dnode/mnode/impl/src/mndShow.c index 6b70825ed4..6e569a04cc 100644 --- a/source/dnode/mnode/impl/src/mndShow.c +++ b/source/dnode/mnode/impl/src/mndShow.c @@ -266,7 +266,7 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) { } size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * pShow->pMeta->numOfColumns + - blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(pBlock); + blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(pBlock->info.numOfCols); SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size); if (pRsp == NULL) { diff --git a/source/libs/executor/src/dataDispatcher.c b/source/libs/executor/src/dataDispatcher.c index eaa366b7e7..b7c7102143 100644 --- a/source/libs/executor/src/dataDispatcher.c +++ b/source/libs/executor/src/dataDispatcher.c @@ -68,10 +68,10 @@ static bool needCompress(const SSDataBlock* pData, int32_t numOfCols) { } // data format: -// +----------------+--------------+----------+--------------------------------------+-------------+-----------+-------------+-----------+ -// |SDataCacheEntry | total length | group id | column#1 length, column#2 length ... | col1 bitmap | col1 data | col2 bitmap | col2 data | .... -// | | (4 bytes) |(8 bytes) | sizeof(int32_t) * numOfCols | actual size | | actual size | | -// +----------------+--------------+----------+--------------------------------------+-------------+-----------+-------------+-----------+ +// +----------------+--------------+----------+--------------------------------------------+--------------------------------------+-------------+-----------+-------------+-----------+ +// |SDataCacheEntry | total length | group id | col1_schema | col2_schema | col3_schema ...| column#1 length, column#2 length ... | col1 bitmap | col1 data | col2 bitmap | col2 data | .... +// | | (4 bytes) |(8 bytes) |(sizeof(int16_t)+sizeof(int32_t))*numOfCols | sizeof(int32_t) * numOfCols | actual size | | 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 static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) { diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 464fb5c957..3a7f1929b9 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -2559,37 +2559,7 @@ int32_t setDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadIn } } - blockDataEnsureCapacity(pBlock, numOfRows); - - int32_t dataLen = *(int32_t*)pStart; - uint64_t groupId = *(uint64_t*)(pStart + sizeof(int32_t)); - pStart += sizeof(int32_t) + sizeof(uint64_t); - - int32_t* colLen = (int32_t*)(pStart); - pStart += sizeof(int32_t) * numOfCols; - - for (int32_t i = 0; i < numOfCols; ++i) { - colLen[i] = htonl(colLen[i]); - ASSERT(colLen[i] >= 0); - - SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i); - if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) { - pColInfoData->varmeta.length = colLen[i]; - pColInfoData->varmeta.allocLen = colLen[i]; - - memcpy(pColInfoData->varmeta.offset, pStart, sizeof(int32_t) * numOfRows); - pStart += sizeof(int32_t) * numOfRows; - - pColInfoData->pData = taosMemoryMalloc(colLen[i]); - } else { - memcpy(pColInfoData->nullbitmap, pStart, BitmapLen(numOfRows)); - pStart += BitmapLen(numOfRows); - } - - memcpy(pColInfoData->pData, pStart, colLen[i]); - pStart += colLen[i]; - } - + blockCompressDecode(pBlock, numOfCols, numOfRows, pStart); // data from mnode relocateColumnData(pRes, pColList, pBlock->pDataBlock); taosArrayDestroy(pBlock->pDataBlock); diff --git a/source/libs/executor/src/tsort.c b/source/libs/executor/src/tsort.c index 7581836d59..310aed8a36 100644 --- a/source/libs/executor/src/tsort.c +++ b/source/libs/executor/src/tsort.c @@ -144,7 +144,9 @@ static int32_t doAddNewExternalMemSource(SDiskbasedBuf *pBuf, SArray* pAllSource (*sourceId) += 1; int32_t rowSize = blockDataGetSerialRowSize(pSource->src.pBlock); - int32_t numOfRows = (getBufPageSize(pBuf) - blockDataGetSerialMetaSize(pBlock))/rowSize; // The value of numOfRows must be greater than 0, which is guaranteed by the previous memory allocation + + // The value of numOfRows must be greater than 0, which is guaranteed by the previous memory allocation + int32_t numOfRows = (getBufPageSize(pBuf) - blockDataGetSerialMetaSize(pBlock->info.numOfCols))/rowSize; ASSERT(numOfRows > 0); return blockDataEnsureCapacity(pSource->src.pBlock, numOfRows); }