Merge pull request #13601 from taosdata/feature/retrievemsg

refactor: do some internal refactor.

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,

include/util/tdef.h

@@ -356,8 +356,8 @@ typedef enum ELogicConditionType {
 
 #define TSDB_DEFAULT_EXPLAIN_VERBOSE false
 
-#define TSDB_EXPLAIN_RESULT_ROW_SIZE    1024
-#define TSDB_EXPLAIN_RESULT_COLUMN_NAME "QUERY PLAN"
+#define TSDB_EXPLAIN_RESULT_ROW_SIZE    512
+#define TSDB_EXPLAIN_RESULT_COLUMN_NAME "QUERY_PLAN"
 
 #define TSDB_MAX_FIELD_LEN             16384
 #define TSDB_MAX_BINARY_LEN            (TSDB_MAX_FIELD_LEN - TSDB_KEYSIZE)  // keep 16384

source/client/src/clientImpl.c

@@ -1339,6 +1339,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;
 

source/client/test/clientTests.cpp

@@ -780,31 +780,28 @@ TEST(testCase, async_api_test) {
 
   taos_query(pConn, "use test");
 
-  TAOS_RES* pRes = taos_query(pConn, "select * from t1");
-
-  taos_query(pConn, "alter table t1 add column b int");
-  pRes = taos_query(pConn, "insert into t1 values(now, 1, 2)");
+  TAOS_RES* pRes = taos_query(pConn, "desc abc1.tu");
   if (taos_errno(pRes) != 0) {
     printf("failed, reason:%s\n", taos_errstr(pRes));
   }
 
-//  int32_t n = 0;
-//  TAOS_ROW    pRow = NULL;
-//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-//  int32_t     numOfFields = taos_num_fields(pRes);
-//
-//  char str[512] = {0};
-//  while ((pRow = taos_fetch_row(pRes)) != NULL) {
-//    int32_t* length = taos_fetch_lengths(pRes);
-//    for(int32_t i = 0; i < numOfFields; ++i) {
-//      printf("(%d):%d " , i, length[i]);
-//    }
-//    printf("\n");
-//
-//    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
-//    printf("%s\n", str);
-//    memset(str, 0, sizeof(str));
-//  }
+  int32_t n = 0;
+  TAOS_ROW    pRow = NULL;
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  int32_t     numOfFields = taos_num_fields(pRes);
+
+  char str[512] = {0};
+  while ((pRow = taos_fetch_row(pRes)) != NULL) {
+    int32_t* length = taos_fetch_lengths(pRes);
+    for(int32_t i = 0; i < numOfFields; ++i) {
+      printf("(%d):%d " , i, length[i]);
+    }
+    printf("\n");
+
+    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+    printf("%s\n", str);
+    memset(str, 0, sizeof(str));
+  }
 
   taos_query_a(pConn, "alter table test.m1 comment 'abcde' ", queryCallback, pConn);
   getchar();

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) {
@@ -1246,7 +1246,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;
 
@@ -1885,34 +1885,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];
@@ -1941,6 +1951,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;
 

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) {

source/libs/command/src/command.c

@@ -28,91 +28,94 @@ static int32_t getSchemaBytes(const SSchema* pSchema) {
   }
 }
 
-// todo : to convert data according to SSDatablock
-static void buildRspData(const STableMeta* pMeta, char* pData) {
-  int32_t* payloadLen = (int32_t*) pData;
-  uint64_t* groupId = (uint64_t*)(pData + sizeof(int32_t));
+static SSDataBlock* buildDescResultDataBlock() {
+  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
+  pBlock->info.numOfCols = DESCRIBE_RESULT_COLS;
+  pBlock->info.hasVarCol = true;
 
-  int32_t* pColSizes = (int32_t*)(pData + sizeof(int32_t) + sizeof(uint64_t));
-  pData = (char*) pColSizes + DESCRIBE_RESULT_COLS * sizeof(int32_t);
+  pBlock->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
 
-  int32_t numOfRows = TABLE_TOTAL_COL_NUM(pMeta);
+  SColumnInfoData infoData = {0};
+  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
+  infoData.info.bytes = DESCRIBE_RESULT_FIELD_LEN;
 
-  // Field
-  int32_t* pOffset = (int32_t*)pData;
-  pData += numOfRows * sizeof(int32_t);
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
+  infoData.info.bytes = DESCRIBE_RESULT_TYPE_LEN;
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  infoData.info.type = TSDB_DATA_TYPE_INT;
+  infoData.info.bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes;;
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
+  infoData.info.bytes = DESCRIBE_RESULT_NOTE_LEN;
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  return pBlock;
+}
+
+static void setDescResultIntoDataBlock(SSDataBlock* pBlock, int32_t numOfRows, STableMeta* pMeta) {
+  blockDataEnsureCapacity(pBlock, numOfRows);
+  pBlock->info.rows = numOfRows;
+
+  // field
+  SColumnInfoData* pCol1 = taosArrayGet(pBlock->pDataBlock, 0);
+  char buf[DESCRIBE_RESULT_FIELD_LEN] = {0};
   for (int32_t i = 0; i < numOfRows; ++i) {
-    STR_TO_VARSTR(pData, pMeta->schema[i].name);
-    int16_t len = varDataTLen(pData);
-    pData += len;
-    *pOffset = pColSizes[0];
-    pOffset += 1;
-    pColSizes[0] += len;
+    STR_TO_VARSTR(buf, pMeta->schema[i].name);
+    colDataAppend(pCol1, i, buf, false);
   }
-  
+
   // Type
-  pOffset = (int32_t*)pData;
-  pData += numOfRows * sizeof(int32_t);
+  SColumnInfoData* pCol2 = taosArrayGet(pBlock->pDataBlock, 1);
   for (int32_t i = 0; i < numOfRows; ++i) {
-    STR_TO_VARSTR(pData, tDataTypes[pMeta->schema[i].type].name);
-    int16_t len = varDataTLen(pData);
-    pData += len;
-    *pOffset = pColSizes[1];
-    pOffset += 1;
-    pColSizes[1] += len;
+    STR_TO_VARSTR(buf, tDataTypes[pMeta->schema[i].type].name);
+    colDataAppend(pCol2, i, buf, false);
   }
 
   // Length
-  pData += BitmapLen(numOfRows);
+  SColumnInfoData* pCol3 = taosArrayGet(pBlock->pDataBlock, 2);
   for (int32_t i = 0; i < numOfRows; ++i) {
-    *(int32_t*)pData = getSchemaBytes(pMeta->schema + i);
-    pData += sizeof(int32_t);
+    int32_t bytes = getSchemaBytes(pMeta->schema + i);
+    colDataAppend(pCol3, i, (const char*)&bytes, false);
   }
-  pColSizes[2] = sizeof(int32_t) * numOfRows;
 
   // Note
-  pOffset = (int32_t*)pData;
-  pData += numOfRows * sizeof(int32_t);
+  SColumnInfoData* pCol4 = taosArrayGet(pBlock->pDataBlock, 3);
   for (int32_t i = 0; i < numOfRows; ++i) {
-    STR_TO_VARSTR(pData, i >= pMeta->tableInfo.numOfColumns ? "TAG" : "");
-    int16_t len = varDataTLen(pData);
-    pData += len;
-    *pOffset = pColSizes[3];
-    pOffset += 1;
-    pColSizes[3] += len;
-  }
-
-  for (int32_t i = 0; i < DESCRIBE_RESULT_COLS; ++i) {
-    pColSizes[i] = htonl(pColSizes[i]);
+    STR_TO_VARSTR(buf, i >= pMeta->tableInfo.numOfColumns ? "TAG" : "");
+    colDataAppend(pCol4, i, buf, false);
   }
-
-
-  *payloadLen = (int32_t)(pData - (char*)payloadLen);
-}
-
-static int32_t calcRspSize(const STableMeta* pMeta) {
-  int32_t numOfRows = TABLE_TOTAL_COL_NUM(pMeta);
-  return sizeof(SRetrieveTableRsp) + 
-      (numOfRows * sizeof(int32_t) + numOfRows * DESCRIBE_RESULT_FIELD_LEN) + 
-      (numOfRows * sizeof(int32_t) + numOfRows * DESCRIBE_RESULT_TYPE_LEN) +
-      (BitmapLen(numOfRows) + numOfRows * sizeof(int32_t)) + 
-      (numOfRows * sizeof(int32_t) + numOfRows * DESCRIBE_RESULT_NOTE_LEN) +
-      sizeof(int32_t) + sizeof(uint64_t);
 }
 
 static int32_t execDescribe(SNode* pStmt, SRetrieveTableRsp** pRsp) {
-  SDescribeStmt* pDesc = (SDescribeStmt*)pStmt;
-  *pRsp = taosMemoryCalloc(1, calcRspSize(pDesc->pMeta));
+  SDescribeStmt* pDesc = (SDescribeStmt*) pStmt;
+  int32_t numOfRows = TABLE_TOTAL_COL_NUM(pDesc->pMeta);
+
+  SSDataBlock* pBlock = buildDescResultDataBlock();
+  setDescResultIntoDataBlock(pBlock, numOfRows, pDesc->pMeta);
+
+  size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
+  *pRsp = taosMemoryCalloc(1, rspSize);
   if (NULL == *pRsp) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
+
   (*pRsp)->useconds = 0;
   (*pRsp)->completed = 1;
   (*pRsp)->precision = 0;
   (*pRsp)->compressed = 0;
   (*pRsp)->compLen = 0;
-  (*pRsp)->numOfRows = htonl(TABLE_TOTAL_COL_NUM(pDesc->pMeta));
-  buildRspData(pDesc->pMeta, (*pRsp)->data);
+  (*pRsp)->numOfRows = htonl(numOfRows);
+  (*pRsp)->numOfCols = htonl(DESCRIBE_RESULT_COLS);
+
+  int32_t len = 0;
+  blockCompressEncode(pBlock, (*pRsp)->data, &len, DESCRIBE_RESULT_COLS, false);
+  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
+
+  blockDataDestroy(pBlock);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/command/src/explain.c

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "tdatablock.h"
 #include "commandInt.h"
 #include "plannodes.h"
 #include "query.h"
@@ -965,9 +966,32 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
     QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
   }
 
-  int32_t colNum = 1;
-  int32_t rspSize = sizeof(SRetrieveTableRsp) + sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t) * colNum +
-                    sizeof(int32_t) * rowNum + pCtx->dataSize;
+  SSDataBlock *pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
+  SColumnInfoData infoData = {0};
+  infoData.info.type  = TSDB_DATA_TYPE_VARCHAR;
+  infoData.info.bytes = TSDB_EXPLAIN_RESULT_ROW_SIZE;
+
+  pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
+  taosArrayPush(pBlock->pDataBlock, &infoData);
+
+  SColumnInfoData* pInfoData = taosArrayGet(pBlock->pDataBlock, 0);
+  pInfoData->hasNull = false;
+  colInfoDataEnsureCapacity(pInfoData, 0, rowNum);
+
+  char buf[1024] = {0};
+  for (int32_t i = 0; i < rowNum; ++i) {
+    SQueryExplainRowInfo *row = taosArrayGet(pCtx->rows, i);
+    varDataCopy(buf, row->buf);
+    ASSERT(varDataTLen(row->buf) == row->len);
+    colDataAppend(pInfoData, i, buf, false);
+  }
+
+  pBlock->info.numOfCols = 1;
+  pBlock->info.rows = rowNum;
+  pBlock->info.hasVarCol = true;
+
+  int32_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
+
   SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)taosMemoryCalloc(1, rspSize);
   if (NULL == rsp) {
     qError("malloc SRetrieveTableRsp failed, size:%d", rspSize);
@@ -977,34 +1001,13 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
   rsp->completed = 1;
   rsp->numOfRows = htonl(rowNum);
 
-  // payload length
-  *(int32_t *)rsp->data =
-      sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t) * colNum + sizeof(int32_t) * rowNum + pCtx->dataSize;
+  int32_t len = 0;
+  blockCompressEncode(pBlock, rsp->data, &len, pBlock->info.numOfCols, 0);
+  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
 
-  // group id
-  *(uint64_t *)(rsp->data + sizeof(int32_t)) = 0;
-
-  // column length
-  int32_t *colLength = (int32_t *)(rsp->data + sizeof(int32_t) + sizeof(uint64_t));
-
-  // varchar column offset segment
-  int32_t *offset = (int32_t *)((char *)colLength + sizeof(int32_t));
-
-  // varchar data real payload
-  char *data = (char *)(offset + rowNum);
-
-  char *start = data;
-  for (int32_t i = 0; i < rowNum; ++i) {
-    SQueryExplainRowInfo *row = taosArrayGet(pCtx->rows, i);
-    offset[i] = data - start;
-
-    varDataCopy(data, row->buf);
-    ASSERT(varDataTLen(row->buf) == row->len);
-    data += row->len;
-  }
+  rsp->compLen = htonl(len);
 
-  *colLength = htonl(data - start);
-  rsp->compLen = htonl(rspSize);
+  blockDataDestroy(pBlock);
 
   *pRsp = rsp;
   return TSDB_CODE_SUCCESS;

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) {

source/libs/executor/src/executorimpl.c

@@ -2571,37 +2571,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);

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);
 }