Skip to content

T
TDengine

taosdata / TDengine
接近 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
未验证 提交 ed3289e0 编写于 作者: D dapan1121 提交者： GitHub
浏览文件
操作

Merge branch '3.0' into feature/3_liaohj

上级 deaa295f 6733636a
隐藏空白更改
内联 并排
Showing with 176 addition and 141 deletion
include/common/tdatablock.h
浏览文件 @ ed3289e0
...@@ -206,7 +206,7 @@ SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int3 ...@@ -206,7 +206,7 @@ SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int3
size_t blockDataGetSize(const SSDataBlock* pBlock); size_t blockDataGetSize(const SSDataBlock* pBlock);
size_t blockDataGetRowSize(SSDataBlock* pBlock); size_t blockDataGetRowSize(SSDataBlock* pBlock);
double blockDataGetSerialRowSize(const 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(SSDataBlock* pDataBlock, SArray* pOrderInfo);
int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst); int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
...@@ -238,7 +238,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool ...@@ -238,7 +238,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool
const char* stbFullName, int32_t vgId); const char* stbFullName, int32_t vgId);
static FORCE_INLINE int32_t blockGetEncodeSize(const SSDataBlock* pBlock) { 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, static FORCE_INLINE int32_t blockCompressColData(SColumnInfoData* pColRes, int32_t numOfRows, char* data,
......
include/util/tdef.h
浏览文件 @ ed3289e0
...@@ -356,8 +356,8 @@ typedef enum ELogicConditionType { ...@@ -356,8 +356,8 @@ typedef enum ELogicConditionType {
#define TSDB_DEFAULT_EXPLAIN_VERBOSE false #define TSDB_DEFAULT_EXPLAIN_VERBOSE false
#define TSDB_EXPLAIN_RESULT_ROW_SIZE 1024 #define TSDB_EXPLAIN_RESULT_ROW_SIZE 512
#define TSDB_EXPLAIN_RESULT_COLUMN_NAME "QUERY PLAN" #define TSDB_EXPLAIN_RESULT_COLUMN_NAME "QUERY_PLAN"
#define TSDB_MAX_FIELD_LEN 16384 #define TSDB_MAX_FIELD_LEN 16384
#define TSDB_MAX_BINARY_LEN (TSDB_MAX_FIELD_LEN - TSDB_KEYSIZE) // keep 16384 #define TSDB_MAX_BINARY_LEN (TSDB_MAX_FIELD_LEN - TSDB_KEYSIZE) // keep 16384
......
source/client/src/clientImpl.c
浏览文件 @ ed3289e0
...@@ -1344,6 +1344,17 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32 ...@@ -1344,6 +1344,17 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
uint64_t groupId = *(uint64_t*)p; uint64_t groupId = *(uint64_t*)p;
p += sizeof(uint64_t); 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; int32_t* colLength = (int32_t*)p;
p += sizeof(int32_t) * numOfCols; p += sizeof(int32_t) * numOfCols;
......
source/client/test/clientTests.cpp
浏览文件 @ ed3289e0
...@@ -778,7 +778,32 @@ TEST(testCase, async_api_test) { ...@@ -778,7 +778,32 @@ TEST(testCase, async_api_test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr); ASSERT_NE(pConn, nullptr);
taos_query_a(pConn, "insert into tm0 values(now()+0s, 1)", queryCallback, pConn); taos_query(pConn, "use test");
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));
}
taos_query_a(pConn, "alter table test.m1 comment 'abcde' ", queryCallback, pConn);
getchar(); getchar();
taos_close(pConn); taos_close(pConn);
} }
......
source/common/src/tdatablock.c
浏览文件 @ ed3289e0
...@@ -682,9 +682,9 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock) { ...@@ -682,9 +682,9 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock) {
* @param pBlock * @param pBlock
* @return * @return
*/ */
size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock) { size_t blockDataGetSerialMetaSize(uint32_t numOfCols) {
// | total rows/total length | block group id | each column length | // | total rows/total length | block group id | column schema | each column length |
return sizeof(int32_t) + sizeof(uint64_t) + pBlock->info.numOfCols * sizeof(int32_t); return sizeof(int32_t) + sizeof(uint64_t) + numOfCols * (sizeof(int16_t) + sizeof(int32_t)) + numOfCols * sizeof(int32_t);
} }
double blockDataGetSerialRowSize(const SSDataBlock* pBlock) { double blockDataGetSerialRowSize(const SSDataBlock* pBlock) {
...@@ -1246,7 +1246,7 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) { ...@@ -1246,7 +1246,7 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) {
} }
size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize) { 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; int32_t rowSize = pBlock->info.rowSize;
...@@ -1885,34 +1885,44 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo ...@@ -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, void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols,
int8_t needCompress) { int8_t needCompress) {
// todo extract method
int32_t* actualLen = (int32_t*)data; int32_t* actualLen = (int32_t*)data;
data += sizeof(int32_t); data += sizeof(int32_t);
uint64_t* groupId = (uint64_t*)data; uint64_t* groupId = (uint64_t*)data;
data += sizeof(uint64_t); 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; int32_t* colSizes = (int32_t*)data;
data += numOfCols * sizeof(int32_t); 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; int32_t numOfRows = pBlock->info.rows;
for (int32_t col = 0; col < numOfCols; ++col) { for (int32_t col = 0; col < numOfCols; ++col) {
SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, col); SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, col);
// copy the null bitmap // copy the null bitmap
size_t metaSize = 0;
if (IS_VAR_DATA_TYPE(pColRes->info.type)) { 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); memcpy(data, pColRes->varmeta.offset, metaSize);
data += metaSize;
(*dataLen) += metaSize;
} else { } else {
int32_t len = BitmapLen(numOfRows); metaSize = BitmapLen(numOfRows);
memcpy(data, pColRes->nullbitmap, len); memcpy(data, pColRes->nullbitmap, metaSize);
data += len;
(*dataLen) += len;
} }
data += metaSize;
(*dataLen) += metaSize;
if (needCompress) { if (needCompress) {
colSizes[col] = blockCompressColData(pColRes, numOfRows, data, needCompress); colSizes[col] = blockCompressColData(pColRes, numOfRows, data, needCompress);
data += colSizes[col]; data += colSizes[col];
...@@ -1941,6 +1951,17 @@ const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t ...@@ -1941,6 +1951,17 @@ const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t
pBlock->info.groupId = *(uint64_t*)pStart; pBlock->info.groupId = *(uint64_t*)pStart;
pStart += sizeof(uint64_t); 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; int32_t* colLen = (int32_t*)pStart;
pStart += sizeof(int32_t) * numOfCols; pStart += sizeof(int32_t) * numOfCols;
......
source/dnode/mnode/impl/src/mndShow.c
浏览文件 @ ed3289e0
...@@ -266,7 +266,7 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) { ...@@ -266,7 +266,7 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
} }
size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * pShow->pMeta->numOfColumns + 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); SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size);
if (pRsp == NULL) { if (pRsp == NULL) {
......
source/libs/command/src/command.c
浏览文件 @ ed3289e0
...@@ -28,91 +28,94 @@ static int32_t getSchemaBytes(const SSchema* pSchema) { ...@@ -28,91 +28,94 @@ static int32_t getSchemaBytes(const SSchema* pSchema) {
} }
} }
// todo : to convert data according to SSDatablock static SSDataBlock* buildDescResultDataBlock() {
static void buildRspData(const STableMeta* pMeta, char* pData) { SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
int32_t* payloadLen = (int32_t*) pData; pBlock->info.numOfCols = DESCRIBE_RESULT_COLS;
uint64_t* groupId = (uint64_t*)(pData + sizeof(int32_t)); pBlock->info.hasVarCol = true;
int32_t* pColSizes = (int32_t*)(pData + sizeof(int32_t) + sizeof(uint64_t)); pBlock->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
pData = (char*) pColSizes + DESCRIBE_RESULT_COLS * sizeof(int32_t);
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 taosArrayPush(pBlock->pDataBlock, &infoData);
int32_t* pOffset = (int32_t*)pData;
pData += numOfRows * sizeof(int32_t); 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) { for (int32_t i = 0; i < numOfRows; ++i) {
STR_TO_VARSTR(pData, pMeta->schema[i].name); STR_TO_VARSTR(buf, pMeta->schema[i].name);
int16_t len = varDataTLen(pData); colDataAppend(pCol1, i, buf, false);
pData += len;
*pOffset = pColSizes[0];
pOffset += 1;
pColSizes[0] += len;
} }
// Type // Type
pOffset = (int32_t*)pData; SColumnInfoData* pCol2 = taosArrayGet(pBlock->pDataBlock, 1);
pData += numOfRows * sizeof(int32_t);
for (int32_t i = 0; i < numOfRows; ++i) { for (int32_t i = 0; i < numOfRows; ++i) {
STR_TO_VARSTR(pData, tDataTypes[pMeta->schema[i].type].name); STR_TO_VARSTR(buf, tDataTypes[pMeta->schema[i].type].name);
int16_t len = varDataTLen(pData); colDataAppend(pCol2, i, buf, false);
pData += len;
*pOffset = pColSizes[1];
pOffset += 1;
pColSizes[1] += len;
} }
// Length // Length
pData += BitmapLen(numOfRows); SColumnInfoData* pCol3 = taosArrayGet(pBlock->pDataBlock, 2);
for (int32_t i = 0; i < numOfRows; ++i) { for (int32_t i = 0; i < numOfRows; ++i) {
*(int32_t*)pData = getSchemaBytes(pMeta->schema + i); int32_t bytes = getSchemaBytes(pMeta->schema + i);
pData += sizeof(int32_t); colDataAppend(pCol3, i, (const char*)&bytes, false);
} }
pColSizes[2] = sizeof(int32_t) * numOfRows;
// Note // Note
pOffset = (int32_t*)pData; SColumnInfoData* pCol4 = taosArrayGet(pBlock->pDataBlock, 3);
pData += numOfRows * sizeof(int32_t);
for (int32_t i = 0; i < numOfRows; ++i) { for (int32_t i = 0; i < numOfRows; ++i) {
STR_TO_VARSTR(pData, i >= pMeta->tableInfo.numOfColumns ? "TAG" : ""); STR_TO_VARSTR(buf, i >= pMeta->tableInfo.numOfColumns ? "TAG" : "");
int16_t len = varDataTLen(pData); colDataAppend(pCol4, i, buf, false);
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]);
} }
*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) { static int32_t execDescribe(SNode* pStmt, SRetrieveTableRsp** pRsp) {
SDescribeStmt* pDesc = (SDescribeStmt*)pStmt; SDescribeStmt* pDesc = (SDescribeStmt*) pStmt;
*pRsp = taosMemoryCalloc(1, calcRspSize(pDesc->pMeta)); 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) { if (NULL == *pRsp) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
(*pRsp)->useconds = 0; (*pRsp)->useconds = 0;
(*pRsp)->completed = 1; (*pRsp)->completed = 1;
(*pRsp)->precision = 0; (*pRsp)->precision = 0;
(*pRsp)->compressed = 0; (*pRsp)->compressed = 0;
(*pRsp)->compLen = 0; (*pRsp)->compLen = 0;
(*pRsp)->numOfRows = htonl(TABLE_TOTAL_COL_NUM(pDesc->pMeta)); (*pRsp)->numOfRows = htonl(numOfRows);
buildRspData(pDesc->pMeta, (*pRsp)->data); (*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; return TSDB_CODE_SUCCESS;
} }
......
source/libs/command/src/explain.c
浏览文件 @ ed3289e0
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "tdatablock.h"
#include "commandInt.h" #include "commandInt.h"
#include "plannodes.h" #include "plannodes.h"
#include "query.h" #include "query.h"
...@@ -965,9 +966,32 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) { ...@@ -965,9 +966,32 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR); QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
} }
int32_t colNum = 1; SSDataBlock *pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
int32_t rspSize = sizeof(SRetrieveTableRsp) + sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t) * colNum + SColumnInfoData infoData = {0};
sizeof(int32_t) * rowNum + pCtx->dataSize; 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); SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)taosMemoryCalloc(1, rspSize);
if (NULL == rsp) { if (NULL == rsp) {
qError("malloc SRetrieveTableRsp failed, size:%d", rspSize); qError("malloc SRetrieveTableRsp failed, size:%d", rspSize);
...@@ -977,34 +1001,13 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) { ...@@ -977,34 +1001,13 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
rsp->completed = 1; rsp->completed = 1;
rsp->numOfRows = htonl(rowNum); rsp->numOfRows = htonl(rowNum);
// payload length int32_t len = 0;
*(int32_t *)rsp->data = blockCompressEncode(pBlock, rsp->data, &len, pBlock->info.numOfCols, 0);
sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t) * colNum + sizeof(int32_t) * rowNum + pCtx->dataSize; ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
// group id rsp->compLen = htonl(len);
*(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;
}
*colLength = htonl(data - start); blockDataDestroy(pBlock);
rsp->compLen = htonl(rspSize);
*pRsp = rsp; *pRsp = rsp;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
......
source/libs/executor/src/dataDispatcher.c
浏览文件 @ ed3289e0
...@@ -68,10 +68,10 @@ static bool needCompress(const SSDataBlock* pData, int32_t numOfCols) { ...@@ -68,10 +68,10 @@ static bool needCompress(const SSDataBlock* pData, int32_t numOfCols) {
} }
// data format: // data format:
// +----------------+--------------+----------+--------------------------------------+-------------+-----------+-------------+-----------+ // +----------------+--------------+----------+--------------------------------------------+--------------------------------------+-------------+-----------+-------------+-----------+
// |SDataCacheEntry | total length | group id | column#1 length, column#2 length ... | col1 bitmap | col1 data | col2 bitmap | col2 data | .... // |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(int32_t) * numOfCols | actual size | | actual size | | // | | (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 // 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 // recorded in the first segment, next to the struct header
static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) { static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) {
......
source/libs/executor/src/executorimpl.c
浏览文件 @ ed3289e0
...@@ -2571,37 +2571,7 @@ int32_t setDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadIn ...@@ -2571,37 +2571,7 @@ int32_t setDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadIn
} }
} }
blockDataEnsureCapacity(pBlock, numOfRows); blockCompressDecode(pBlock, numOfCols, numOfRows, pStart);
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];
}
// data from mnode // data from mnode
relocateColumnData(pRes, pColList, pBlock->pDataBlock); relocateColumnData(pRes, pColList, pBlock->pDataBlock);
taosArrayDestroy(pBlock->pDataBlock); taosArrayDestroy(pBlock->pDataBlock);
......
source/libs/executor/src/tsort.c
浏览文件 @ ed3289e0
...@@ -144,7 +144,9 @@ static int32_t doAddNewExternalMemSource(SDiskbasedBuf *pBuf, SArray* pAllSource ...@@ -144,7 +144,9 @@ static int32_t doAddNewExternalMemSource(SDiskbasedBuf *pBuf, SArray* pAllSource
(*sourceId) += 1; (*sourceId) += 1;
int32_t rowSize = blockDataGetSerialRowSize(pSource->src.pBlock); 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); ASSERT(numOfRows > 0);
return blockDataEnsureCapacity(pSource->src.pBlock, numOfRows); return blockDataEnsureCapacity(pSource->src.pBlock, numOfRows);
} }
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或