xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #10263 from taosdata/feature/privilege 

serialize msg
This commit is contained in:
Shengliang Guan 2022-02-15 17:05:37 +08:00 committed by GitHub
parent b6625ac4af cfa76f8329
commit f9644ffd35
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
27 changed files with 929 additions and 724 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
include/common/tmsg.h
View File

@ -803,6 +803,9 @@ typedef struct {
char tbName[TSDB_TABLE_NAME_LEN];
} STableInfoReq;
int32_t tSerializeSTableInfoReq(void* buf, int32_t bufLen, STableInfoReq* pReq);
int32_t tDeserializeSTableInfoReq(void* buf, int32_t bufLen, STableInfoReq* pReq);
typedef struct {
int8_t metaClone; // create local clone of the cached table meta
int32_t numOfVgroups;
@ -839,9 +842,21 @@ typedef struct {
uint64_t suid;
uint64_t tuid;
int32_t vgId;
SSchema pSchema[];
SSchema* pSchemas;
} STableMetaRsp;
int32_t tSerializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
void tFreeSTableMetaRsp(STableMetaRsp* pRsp);
typedef struct {
SArray* pArray; // Array of STableMetaRsp
} STableMetaBatchRsp;
int32_t tSerializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp);
int32_t tDeserializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp);
void tFreeSTableMetaBatchRsp(STableMetaBatchRsp* pRsp);
typedef struct {
int32_t numOfTables;
int32_t numOfVgroup;
@ -875,18 +890,15 @@ int32_t tSerializeSShowReq(void* buf, int32_t bufLen, SShowReq* pReq);
int32_t tDeserializeSShowReq(void* buf, int32_t bufLen, SShowReq* pReq);
void tFreeSShowReq(SShowReq* pReq);
typedef struct {
char db[TSDB_DB_FNAME_LEN];
int32_t numOfVgroup;
int32_t vgid[];
} SCompactReq;
typedef struct {
int64_t showId;
STableMetaRsp tableMeta;
} SShowRsp;
} SShowRsp, SVShowTablesRsp;
int32_t tSerializeSShowRsp(void* buf, int32_t bufLen, SShowRsp* pRsp);
int32_t tDeserializeSShowRsp(void* buf, int32_t bufLen, SShowRsp* pRsp);
void tFreeSShowRsp(SShowRsp* pRsp);
// todo: the show handle should be replaced with id
typedef struct {
int64_t showId;
int8_t free;
@ -1414,11 +1426,6 @@ typedef struct {
SMsgHead head;
} SVShowTablesReq;
typedef struct {
int64_t id;
STableMetaRsp metaInfo;
} SVShowTablesRsp;
typedef struct {
SMsgHead head;
int32_t id;
@ -1623,7 +1630,7 @@ int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchR
static FORCE_INLINE int32_t tEncodeSKv(SCoder* pEncoder, const SKv* pKv) {
if (tEncodeI32(pEncoder, pKv->key) < 0) return -1;
if (tEncodeI32(pEncoder, pKv->valueLen) < 0) return -1;
if (tEncodeCStrWithLen(pEncoder, (const char*)pKv->value, pKv->valueLen) < 0) return -1;
if (tEncodeBinary(pEncoder, (const char*)pKv->value, pKv->valueLen) < 0) return -1;
return 0;
}
@ -1839,7 +1846,7 @@ typedef struct {
SSchema* pSchema;
} SSchemaWrapper;
static FORCE_INLINE int32_t tEncodeSSchema(void** buf, const SSchema* pSchema) {
static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
int32_t tlen = 0;
tlen += taosEncodeFixedI8(buf, pSchema->type);
tlen += taosEncodeFixedI32(buf, pSchema->bytes);
@ -1848,7 +1855,7 @@ static FORCE_INLINE int32_t tEncodeSSchema(void** buf, const SSchema* pSchema) {
return tlen;
}
static FORCE_INLINE void* tDecodeSSchema(void* buf, SSchema* pSchema) {
static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
buf = taosDecodeFixedI8(buf, &pSchema->type);
buf = taosDecodeFixedI32(buf, &pSchema->bytes);
buf = taosDecodeFixedI32(buf, &pSchema->colId);
@ -1856,11 +1863,27 @@ static FORCE_INLINE void* tDecodeSSchema(void* buf, SSchema* pSchema) {
return buf;
}
static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) {
if (tEncodeI8(pEncoder, pSchema->type) < 0) return -1;
if (tEncodeI32(pEncoder, pSchema->bytes) < 0) return -1;
if (tEncodeI32(pEncoder, pSchema->colId) < 0) return -1;
if (tEncodeCStr(pEncoder, pSchema->name) < 0) return -1;
return 0;
}
static FORCE_INLINE int32_t tDecodeSSchema(SCoder* pDecoder, SSchema* pSchema) {
if (tDecodeI8(pDecoder, &pSchema->type) < 0) return -1;
if (tDecodeI32(pDecoder, &pSchema->bytes) < 0) return -1;
if (tDecodeI32(pDecoder, &pSchema->colId) < 0) return -1;
if (tDecodeCStrTo(pDecoder, pSchema->name) < 0) return -1;
return 0;
}
static FORCE_INLINE int32_t tEncodeSSchemaWrapper(void** buf, const SSchemaWrapper* pSW) {
int32_t tlen = 0;
tlen += taosEncodeFixedU32(buf, pSW->nCols);
for (int32_t i = 0; i < pSW->nCols; i++) {
tlen += tEncodeSSchema(buf, &pSW->pSchema[i]);
tlen += taosEncodeSSchema(buf, &pSW->pSchema[i]);
}
return tlen;
}
@ -1871,8 +1894,9 @@ static FORCE_INLINE void* tDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW)
if (pSW->pSchema == NULL) {
return NULL;
}
for (int32_t i = 0; i < pSW->nCols; i++) {
buf = tDecodeSSchema(buf, &pSW->pSchema[i]);
buf = taosDecodeSSchema(buf, &pSW->pSchema[i]);
}
return buf;
}

52
source/client/src/clientHb.c
View File

@ -70,62 +70,42 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
}
}
tFreeSUseDbBatchRsp(&batchUseRsp);
return TSDB_CODE_SUCCESS;
}
static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
int32_t msgLen = 0;
int32_t code = 0;
int32_t schemaNum = 0;
while (msgLen < valueLen) {
STableMetaRsp *rsp = (STableMetaRsp *)((char *)value + msgLen);
rsp->numOfColumns = ntohl(rsp->numOfColumns);
rsp->suid = be64toh(rsp->suid);
rsp->dbId = be64toh(rsp->dbId);
STableMetaBatchRsp batchMetaRsp = {0};
if (tDeserializeSTableMetaBatchRsp(value, valueLen, &batchMetaRsp) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
int32_t numOfBatchs = taosArrayGetSize(batchMetaRsp.pArray);
for (int32_t i = 0; i < numOfBatchs; ++i) {
STableMetaRsp *rsp = taosArrayGet(batchMetaRsp.pArray, i);
if (rsp->numOfColumns < 0) {
schemaNum = 0;
tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
catalogRemoveStbMeta(pCatalog, rsp->dbFName, rsp->dbId, rsp->stbName, rsp->suid);
} else {
tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
rsp->numOfTags = ntohl(rsp->numOfTags);
rsp->sversion = ntohl(rsp->sversion);
rsp->tversion = ntohl(rsp->tversion);
rsp->tuid = be64toh(rsp->tuid);
rsp->vgId = ntohl(rsp->vgId);
SSchema* pSchema = rsp->pSchema;
schemaNum = rsp->numOfColumns + rsp->numOfTags;
for (int i = 0; i < schemaNum; ++i) {
pSchema->bytes = ntohl(pSchema->bytes);
pSchema->colId = ntohl(pSchema->colId);
pSchema++;
}
if (rsp->pSchema[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
tscError("invalid colId[%d] for the first column in table meta rsp msg", rsp->pSchema[0].colId);
if (rsp->pSchemas[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
tscError("invalid colId[%d] for the first column in table meta rsp msg", rsp->pSchemas[0].colId);
tFreeSTableMetaBatchRsp(&batchMetaRsp);
return TSDB_CODE_TSC_INVALID_VALUE;
}
}
catalogUpdateSTableMeta(pCatalog, rsp);
}
msgLen += sizeof(STableMetaRsp) + schemaNum * sizeof(SSchema);
}
tFreeSTableMetaBatchRsp(&batchMetaRsp);
return TSDB_CODE_SUCCESS;
}
static int32_t hbQueryHbRspHandle(struct SAppHbMgr *pAppHbMgr, SClientHbRsp* pRsp) {
SHbConnInfo * info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey));
if (NULL == info) {

40
source/client/src/clientMsgHandler.c
View File

@ -20,14 +20,14 @@
#include "clientLog.h"
#include "catalog.h"
int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
int32_t (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
static void setErrno(SRequestObj* pRequest, int32_t code) {
pRequest->code = code;
terrno = code;
}
int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
int32_t genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
setErrno(pRequest, code);
@ -36,7 +36,7 @@ int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
return code;
}
int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
free(pMsg->pData);
@ -61,9 +61,9 @@ int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &pConnect->epSet);
}
for (int i = 0; i < pConnect->epSet.numOfEps; ++i) {
tscDebug("0x%" PRIx64 " epSet.fqdn[%d]:%s port:%d, connObj:0x%"PRIx64, pRequest->requestId, i, pConnect->epSet.eps[i].fqdn,
pConnect->epSet.eps[i].port, pTscObj->id);
for (int32_t i = 0; i < pConnect->epSet.numOfEps; ++i) {
tscDebug("0x%" PRIx64 " epSet.fqdn[%d]:%s port:%d, connObj:0x%" PRIx64, pRequest->requestId, i,
pConnect->epSet.eps[i].fqdn, pConnect->epSet.eps[i].port, pTscObj->id);
}
pTscObj->connId = pConnect->connId;
@ -135,39 +135,29 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
return code;
}
SShowRsp* pShow = (SShowRsp *)pMsg->pData;
pShow->showId = htobe64(pShow->showId);
SShowRsp showRsp = {0};
tDeserializeSShowRsp(pMsg->pData, pMsg->len, &showRsp);
STableMetaRsp *pMetaMsg = &showRsp.tableMeta;
STableMetaRsp *pMetaMsg = &(pShow->tableMeta);
pMetaMsg->numOfColumns = htonl(pMetaMsg->numOfColumns);
SSchema* pSchema = pMetaMsg->pSchema;
pMetaMsg->tuid = htobe64(pMetaMsg->tuid);
for (int i = 0; i < pMetaMsg->numOfColumns; ++i) {
pSchema->bytes = htonl(pSchema->bytes);
pSchema->colId = htonl(pSchema->colId);
pSchema++;
}
pSchema = pMetaMsg->pSchema;
tfree(pRequest->body.resInfo.pRspMsg);
pRequest->body.resInfo.pRspMsg = pMsg->pData;
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
if (pResInfo->fields == NULL) {
TAOS_FIELD* pFields = calloc(pMetaMsg->numOfColumns, sizeof(TAOS_FIELD));
for (int32_t i = 0; i < pMetaMsg->numOfColumns; ++i) {
tstrncpy(pFields[i].name, pSchema[i].name, tListLen(pFields[i].name));
pFields[i].type = pSchema[i].type;
pFields[i].bytes = pSchema[i].bytes;
SSchema* pSchema = &pMetaMsg->pSchemas[i];
tstrncpy(pFields[i].name, pSchema->name, tListLen(pFields[i].name));
pFields[i].type = pSchema->type;
pFields[i].bytes = pSchema->bytes;
}
pResInfo->fields = pFields;
}
pResInfo->numOfCols = pMetaMsg->numOfColumns;
pRequest->body.showInfo.execId = pShow->showId;
pRequest->body.showInfo.execId = showRsp.showId;
tFreeSShowRsp(&showRsp);
// todo
if (pRequest->type == TDMT_VND_SHOW_TABLES) {

212
source/common/src/tmsg.c
View File

@ -1485,7 +1485,7 @@ int32_t tSerializeSShowReq(void *buf, int32_t bufLen, SShowReq *pReq) {
if (tEncodeCStr(&encoder, pReq->db) < 0) return -1;
if (tEncodeI32(&encoder, pReq->payloadLen) < 0) return -1;
if (pReq->payloadLen > 0) {
if (tEncodeCStr(&encoder, pReq->payload) < 0) return -1;
if (tEncodeBinary(&encoder, pReq->payload, pReq->payloadLen) < 0) return -1;
}
tEndEncode(&encoder);
@ -1513,7 +1513,7 @@ int32_t tDeserializeSShowReq(void *buf, int32_t bufLen, SShowReq *pReq) {
return 0;
}
void tFreeSShowReq(SShowReq *pReq) { free(pReq->payload); }
void tFreeSShowReq(SShowReq *pReq) { tfree(pReq->payload); }
int32_t tSerializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableReq *pReq) {
SCoder encoder = {0};
@ -1541,3 +1541,211 @@ int32_t tDeserializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableR
tCoderClear(&decoder);
return 0;
}
static int32_t tEncodeSTableMetaRsp(SCoder *pEncoder, STableMetaRsp *pRsp) {
if (tEncodeCStr(pEncoder, pRsp->tbName) < 0) return -1;
if (tEncodeCStr(pEncoder, pRsp->stbName) < 0) return -1;
if (tEncodeCStr(pEncoder, pRsp->dbFName) < 0) return -1;
if (tEncodeU64(pEncoder, pRsp->dbId) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->numOfTags) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->numOfColumns) < 0) return -1;
if (tEncodeI8(pEncoder, pRsp->precision) < 0) return -1;
if (tEncodeI8(pEncoder, pRsp->tableType) < 0) return -1;
if (tEncodeI8(pEncoder, pRsp->update) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->sversion) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->tversion) < 0) return -1;
if (tEncodeU64(pEncoder, pRsp->suid) < 0) return -1;
if (tEncodeU64(pEncoder, pRsp->tuid) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->vgId) < 0) return -1;
for (int32_t i = 0; i < pRsp->numOfColumns + pRsp->numOfTags; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i];
if (tEncodeSSchema(pEncoder, pSchema) < 0) return -1;
}
return 0;
}
static int32_t tDecodeSTableMetaRsp(SCoder *pDecoder, STableMetaRsp *pRsp) {
if (tDecodeCStrTo(pDecoder, pRsp->tbName) < 0) return -1;
if (tDecodeCStrTo(pDecoder, pRsp->stbName) < 0) return -1;
if (tDecodeCStrTo(pDecoder, pRsp->dbFName) < 0) return -1;
if (tDecodeU64(pDecoder, &pRsp->dbId) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->numOfTags) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->numOfColumns) < 0) return -1;
if (tDecodeI8(pDecoder, &pRsp->precision) < 0) return -1;
if (tDecodeI8(pDecoder, &pRsp->tableType) < 0) return -1;
if (tDecodeI8(pDecoder, &pRsp->update) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->sversion) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->tversion) < 0) return -1;
if (tDecodeU64(pDecoder, &pRsp->suid) < 0) return -1;
if (tDecodeU64(pDecoder, &pRsp->tuid) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->vgId) < 0) return -1;
int32_t totalCols = pRsp->numOfTags + pRsp->numOfColumns;
pRsp->pSchemas = malloc(sizeof(SSchema) * totalCols);
if (pRsp->pSchemas == NULL) return -1;
for (int32_t i = 0; i < totalCols; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i];
if (tDecodeSSchema(pDecoder, pSchema) < 0) return -1;
}
return 0;
}
int32_t tSerializeSTableMetaRsp(void *buf, int32_t bufLen, STableMetaRsp *pRsp) {
SCoder encoder = {0};
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeSTableMetaRsp(&encoder, pRsp) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tCoderClear(&encoder);
return tlen;
}
int32_t tSerializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatchRsp *pRsp) {
SCoder encoder = {0};
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
if (tStartEncode(&encoder) < 0) return -1;
int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
if (tEncodeI32(&encoder, numOfBatch) < 0) return -1;
for (int32_t i = 0; i < numOfBatch; ++i) {
STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pArray, i);
if (tEncodeSTableMetaRsp(&encoder, pMetaRsp) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tCoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSTableMetaRsp(void *buf, int32_t bufLen, STableMetaRsp *pRsp) {
SCoder decoder = {0};
tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeSTableMetaRsp(&decoder, pRsp) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
return 0;
}
int32_t tDeserializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatchRsp *pRsp) {
SCoder decoder = {0};
tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
if (tStartDecode(&decoder) < 0) return -1;
int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
if (tDecodeI32(&decoder, &numOfBatch) < 0) return -1;
pRsp->pArray = taosArrayInit(numOfBatch, sizeof(STableMetaRsp));
if (pRsp->pArray == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < numOfBatch; ++i) {
STableMetaRsp tableMetaRsp = {0};
if (tDecodeSTableMetaRsp(&decoder, &tableMetaRsp) < 0) return -1;
taosArrayPush(pRsp->pArray, &tableMetaRsp);
}
tEndDecode(&decoder);
tCoderClear(&decoder);
return 0;
}
void tFreeSTableMetaRsp(STableMetaRsp *pRsp) { tfree(pRsp->pSchemas); }
void tFreeSTableMetaBatchRsp(STableMetaBatchRsp *pRsp) {
int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
for (int32_t i = 0; i < numOfBatch; ++i) {
STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pArray, i);
tFreeSTableMetaRsp(pMetaRsp);
}
taosArrayDestroy(pRsp->pArray);
}
int32_t tSerializeSShowRsp(void *buf, int32_t bufLen, SShowRsp *pRsp) {
SCoder encoder = {0};
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->showId) < 0) return -1;
if (tEncodeSTableMetaRsp(&encoder, &pRsp->tableMeta) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tCoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSShowRsp(void *buf, int32_t bufLen, SShowRsp *pRsp) {
SCoder decoder = {0};
tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI64(&decoder, &pRsp->showId) < 0) return -1;
if (tDecodeSTableMetaRsp(&decoder, &pRsp->tableMeta) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
return 0;
}
void tFreeSShowRsp(SShowRsp *pRsp) { tFreeSTableMetaRsp(&pRsp->tableMeta); }
int32_t tSerializeSTableInfoReq(void *buf, int32_t bufLen, STableInfoReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
if (buf != NULL) {
buf = (char *)buf + headLen;
bufLen -= headLen;
}
SCoder encoder = {0};
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->dbFName) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->tbName) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tCoderClear(&encoder);
if (buf != NULL) {
SMsgHead *pHead = (SMsgHead *)((char *)buf - headLen);
pHead->vgId = htonl(pReq->header.vgId);
pHead->contLen = htonl(tlen + headLen);
}
return tlen + headLen;
}
int32_t tDeserializeSTableInfoReq(void *buf, int32_t bufLen, STableInfoReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
SMsgHead *pHead = buf;
pHead->vgId = pReq->header.vgId;
pHead->contLen = pReq->header.contLen;
SCoder decoder = {0};
tCoderInit(&decoder, TD_LITTLE_ENDIAN, (char *)buf + headLen, bufLen - headLen, TD_DECODER);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->dbFName) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->tbName) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
return 0;
}

2
source/dnode/mgmt/impl/test/sut/inc/sut.h
View File

@ -74,7 +74,7 @@ class Testbase {
private:
int64_t showId;
STableMetaRsp* pMeta;
STableMetaRsp metaRsp;
SRetrieveTableRsp* pRetrieveRsp;
char* pData;
int32_t pos;

47
source/dnode/mgmt/impl/test/sut/src/sut.cpp
View File

@ -56,9 +56,16 @@ void Testbase::Init(const char* path, int16_t port) {
server.Start(path, fqdn, port, firstEp);
client.Init("root", "taosdata", fqdn, port);
taosMsleep(1100);
tFreeSTableMetaRsp(&metaRsp);
showId = 0;
pData = 0;
pos = 0;
pRetrieveRsp = NULL;
}
void Testbase::Cleanup() {
tFreeSTableMetaRsp(&metaRsp);
server.Stop();
client.Cleanup();
dndCleanup();
@ -85,51 +92,43 @@ void Testbase::SendShowMetaReq(int8_t showType, const char* db) {
strcpy(showReq.db, db);
int32_t contLen = tSerializeSShowReq(NULL, 0, &showReq);
char* pReq = (char*)rpcMallocCont(contLen);
void* pReq = rpcMallocCont(contLen);
tSerializeSShowReq(pReq, contLen, &showReq);
tFreeSShowReq(&showReq);
SRpcMsg* pRsp = SendReq(TDMT_MND_SHOW, pReq, contLen);
SShowRsp* pShowRsp = (SShowRsp*)pRsp->pCont;
SRpcMsg* pRsp = SendReq(TDMT_MND_SHOW, pReq, contLen);
ASSERT(pRsp->pCont != nullptr);
ASSERT(pShowRsp != nullptr);
pShowRsp->showId = htobe64(pShowRsp->showId);
pMeta = &pShowRsp->tableMeta;
pMeta->numOfTags = htonl(pMeta->numOfTags);
pMeta->numOfColumns = htonl(pMeta->numOfColumns);
pMeta->sversion = htonl(pMeta->sversion);
pMeta->tversion = htonl(pMeta->tversion);
pMeta->tuid = htobe64(pMeta->tuid);
pMeta->suid = htobe64(pMeta->suid);
showId = pShowRsp->showId;
SShowRsp showRsp = {0};
tDeserializeSShowRsp(pRsp->pCont, pRsp->contLen, &showRsp);
tFreeSTableMetaRsp(&metaRsp);
metaRsp = showRsp.tableMeta;
showId = showRsp.showId;
}
int32_t Testbase::GetMetaColId(int32_t index) {
SSchema* pSchema = &pMeta->pSchema[index];
pSchema->colId = htonl(pSchema->colId);
SSchema* pSchema = &metaRsp.pSchemas[index];
return pSchema->colId;
}
int8_t Testbase::GetMetaType(int32_t index) {
SSchema* pSchema = &pMeta->pSchema[index];
SSchema* pSchema = &metaRsp.pSchemas[index];
return pSchema->type;
}
int32_t Testbase::GetMetaBytes(int32_t index) {
SSchema* pSchema = &pMeta->pSchema[index];
pSchema->bytes = htonl(pSchema->bytes);
SSchema* pSchema = &metaRsp.pSchemas[index];
return pSchema->bytes;
}
const char* Testbase::GetMetaName(int32_t index) {
SSchema* pSchema = &pMeta->pSchema[index];
SSchema* pSchema = &metaRsp.pSchemas[index];
return pSchema->name;
}
int32_t Testbase::GetMetaNum() { return pMeta->numOfColumns; }
int32_t Testbase::GetMetaNum() { return metaRsp.numOfColumns; }
const char* Testbase::GetMetaTbName() { return pMeta->tbName; }
const char* Testbase::GetMetaTbName() { return metaRsp.tbName; }
void Testbase::SendShowRetrieveReq() {
SRetrieveTableReq retrieveReq = {0};
@ -150,7 +149,7 @@ void Testbase::SendShowRetrieveReq() {
pos = 0;
}
const char* Testbase::GetShowName() { return pMeta->tbName; }
const char* Testbase::GetShowName() { return metaRsp.tbName; }
int8_t Testbase::GetShowInt8() {
int8_t data = *((int8_t*)(pData + pos));
@ -191,6 +190,6 @@ const char* Testbase::GetShowBinary(int32_t len) {
int32_t Testbase::GetShowRows() { return pRetrieveRsp->numOfRows; }
STableMetaRsp* Testbase::GetShowMeta() { return pMeta; }
STableMetaRsp* Testbase::GetShowMeta() { return &metaRsp; }
SRetrieveTableRsp* Testbase::GetRetrieveRsp() { return pRetrieveRsp; }

2
source/dnode/mnode/impl/inc/mndDb.h
View File

@ -24,7 +24,7 @@ extern "C" {
int32_t mndInitDb(SMnode *pMnode);
void mndCleanupDb(SMnode *pMnode);
SDbObj *mndAcquireDb(SMnode *pMnode, char *db);
SDbObj *mndAcquireDb(SMnode *pMnode, const char *db);
void mndReleaseDb(SMnode *pMnode, SDbObj *pDb);
int32_t mndValidateDbInfo(SMnode *pMnode, SDbVgVersion *pDbs, int32_t numOfDbs, void **ppRsp, int32_t *pRspLen);

12
source/dnode/mnode/impl/inc/mndStb.h
View File

@ -22,14 +22,12 @@
extern "C" {
#endif
int32_t mndInitStb(SMnode *pMnode);
void mndCleanupStb(SMnode *pMnode);
int32_t mndInitStb(SMnode *pMnode);
void mndCleanupStb(SMnode *pMnode);
SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName);
void mndReleaseStb(SMnode *pMnode, SStbObj *pStb);
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *stbs, int32_t num, void **rsp, int32_t *rspLen);
void mndReleaseStb(SMnode *pMnode, SStbObj *pStb);
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *pStbs, int32_t numOfStbs, void **ppRsp,
int32_t *pRspLen);
#ifdef __cplusplus
}

10
source/dnode/mnode/impl/src/mndBnode.c
View File

@ -433,27 +433,27 @@ static int32_t mndGetBnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_EP_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "endpoint");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

10
source/dnode/mnode/impl/src/mndCluster.c
View File

@ -165,27 +165,27 @@ static int32_t mndCreateDefaultCluster(SMnode *pMnode) {
static int32_t mndGetClusterMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta) {
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_BIGINT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_CLUSTER_ID_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
strcpy(pMeta->tbName, mndShowStr(pShow->type));
pShow->numOfColumns = cols;

42
source/dnode/mnode/impl/src/mndDb.c
View File

@ -194,7 +194,7 @@ static int32_t mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew) {
return 0;
}
SDbObj *mndAcquireDb(SMnode *pMnode, char *db) {
SDbObj *mndAcquireDb(SMnode *pMnode, const char *db) {
SSdb *pSdb = pMnode->pSdb;
SDbObj *pDb = sdbAcquire(pSdb, SDB_DB, db);
if (pDb == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
@ -1111,117 +1111,117 @@ static int32_t mndGetDbMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMe
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = (TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "vgroups");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "ntables");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "replica");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "quorum");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "days");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 24 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "keep0,keep1,keep2");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "cache");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "blocks");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "minrows");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "maxrows");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 1;
pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
strcpy(pSchema[cols].name, "wallevel");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "fsync");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 1;
pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
strcpy(pSchema[cols].name, "comp");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 1;
pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
strcpy(pSchema[cols].name, "cachelast");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 3 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "precision");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 1;
pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
strcpy(pSchema[cols].name, "update");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

26
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -628,21 +628,21 @@ static int32_t mndProcessConfigDnodeRsp(SMnodeMsg *pRsp) {
static int32_t mndGetConfigMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
tstrncpy(pSchema[cols].name, "name", sizeof(pSchema[cols].name));
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_CONIIG_VALUE_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
tstrncpy(pSchema[cols].name, "value", sizeof(pSchema[cols].name));
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
@ -705,51 +705,51 @@ static int32_t mndGetDnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_EP_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "endpoint");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "vnodes");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "support_vnodes");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 10 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "status");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 24 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "offline_reason");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

18
source/dnode/mnode/impl/src/mndFunc.c
View File

@ -468,51 +468,51 @@ static int32_t mndGetFuncMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *p
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_FUNC_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = PATH_MAX + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "comment");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "aggregate");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_TYPE_STR_MAX_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "outputtype");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "code_len");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "bufsize");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

14
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -620,39 +620,39 @@ static int32_t mndGetMnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_EP_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "endpoint");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 12 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "role");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "role_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

50
source/dnode/mnode/impl/src/mndProfile.c
View File

@ -578,53 +578,53 @@ static int32_t mndGetConnsMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
mndReleaseUser(pMnode, pUser);
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "connId");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_USER_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "user");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
// app name
pShow->bytes[cols] = TSDB_APP_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "program");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
// app pid
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "pid");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ip:port");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "login_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "last_access");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
@ -707,93 +707,93 @@ static int32_t mndGetQueryMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
mndReleaseUser(pMnode, pUser);
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "queryId");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "connId");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_USER_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "user");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ip:port");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 22 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "qid");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "created_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_BIGINT;
strcpy(pSchema[cols].name, "time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = QUERY_OBJ_ID_SIZE + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sql_obj_id");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "pid");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_EP_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ep");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 1;
pSchema[cols].type = TSDB_DATA_TYPE_BOOL;
strcpy(pSchema[cols].name, "stable_query");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "sub_queries");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_SHOW_SUBQUERY_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sub_query_info");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sql");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

10
source/dnode/mnode/impl/src/mndQnode.c
View File

@ -433,27 +433,27 @@ static int32_t mndGetQnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_EP_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "endpoint");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

21
source/dnode/mnode/impl/src/mndShow.c
View File

@ -120,6 +120,7 @@ static int32_t mndProcessShowReq(SMnodeMsg *pReq) {
SShowMgmt *pMgmt = &pMnode->showMgmt;
int32_t code = -1;
SShowReq showReq = {0};
SShowRsp showRsp = {0};
if (tDeserializeSShowReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &showReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
@ -142,25 +143,26 @@ static int32_t mndProcessShowReq(SMnodeMsg *pReq) {
goto SHOW_OVER;
}
int32_t size = sizeof(SShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE;
SShowRsp *pRsp = rpcMallocCont(size);
if (pRsp == NULL) {
showRsp.showId = pShow->id;
showRsp.tableMeta.pSchemas = calloc(TSDB_MAX_COLUMNS, sizeof(SSchema));
if (showRsp.tableMeta.pSchemas == NULL) {
mndReleaseShowObj(pShow, true);
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto SHOW_OVER;
}
code = (*metaFp)(pReq, pShow, &pRsp->tableMeta);
code = (*metaFp)(pReq, pShow, &showRsp.tableMeta);
mDebug("show:0x%" PRIx64 ", get meta finished, numOfRows:%d cols:%d showReq.type:%s, result:%s", pShow->id,
pShow->numOfRows, pShow->numOfColumns, mndShowStr(showReq.type), tstrerror(code));
if (code == TSDB_CODE_SUCCESS) {
pReq->contLen = sizeof(SShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
pReq->pCont = pRsp;
pRsp->showId = htobe64(pShow->id);
if (code == 0) {
int32_t bufLen = tSerializeSShowRsp(NULL, 0, &showRsp);
void *pBuf = rpcMallocCont(bufLen);
tSerializeSShowRsp(pBuf, bufLen, &showRsp);
pReq->contLen = bufLen;
pReq->pCont = pBuf;
mndReleaseShowObj(pShow, false);
} else {
rpcFreeCont(pRsp);
mndReleaseShowObj(pShow, true);
}
@ -170,6 +172,7 @@ SHOW_OVER:
}
tFreeSShowReq(&showReq);
tFreeSShowRsp(&showRsp);
return code;
}

10
source/dnode/mnode/impl/src/mndSnode.c
View File

@ -436,27 +436,27 @@ static int32_t mndGetSnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_EP_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "endpoint");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

310
source/dnode/mnode/impl/src/mndStb.c
View File

@ -1179,19 +1179,59 @@ static int32_t mndProcessVDropStbRsp(SMnodeMsg *pRsp) {
return 0;
}
static int32_t mndProcessStbMetaReq(SMnodeMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
STableInfoReq *pInfo = pReq->rpcMsg.pCont;
static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableMetaRsp *pRsp) {
taosRLockLatch(&pStb->lock);
int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
pRsp->pSchemas = calloc(totalCols, sizeof(SSchema));
if (pRsp->pSchemas == NULL) {
taosRUnLockLatch(&pStb->lock);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
strcpy(pRsp->dbFName, pStb->db);
strcpy(pRsp->tbName, tbName);
strcpy(pRsp->stbName, tbName);
pRsp->dbId = pDb->uid;
pRsp->numOfTags = pStb->numOfTags;
pRsp->numOfColumns = pStb->numOfColumns;
pRsp->precision = pDb->cfg.precision;
pRsp->tableType = TSDB_SUPER_TABLE;
pRsp->update = pDb->cfg.update;
pRsp->sversion = pStb->version;
pRsp->suid = pStb->uid;
pRsp->tuid = pStb->uid;
for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i];
SSchema *pSrcSchema = &pStb->pColumns[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = pSrcSchema->colId;
pSchema->bytes = pSrcSchema->bytes;
}
for (int32_t i = 0; i < pStb->numOfTags; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
SSchema *pSrcSchema = &pStb->pTags[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = pSrcSchema->colId;
pSchema->bytes = pSrcSchema->bytes;
}
taosRUnLockLatch(&pStb->lock);
return 0;
}
static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) {
char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
snprintf(tbFName, sizeof(tbFName), "%s.%s", pInfo->dbFName, pInfo->tbName);
snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
mDebug("stb:%s, start to retrieve meta", tbFName);
SDbObj *pDb = mndAcquireDb(pMnode, pInfo->dbFName);
SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
if (pDb == NULL) {
terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
mError("stb:%s, failed to retrieve meta since %s", tbFName, terrstr());
return -1;
}
@ -1199,175 +1239,103 @@ static int32_t mndProcessStbMetaReq(SMnodeMsg *pReq) {
if (pStb == NULL) {
mndReleaseDb(pMnode, pDb);
terrno = TSDB_CODE_MND_INVALID_STB;
mError("stb:%s, failed to get meta since %s", tbFName, terrstr());
return -1;
}
taosRLockLatch(&pStb->lock);
int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
int32_t contLen = sizeof(STableMetaRsp) + totalCols * sizeof(SSchema);
STableMetaRsp *pMeta = rpcMallocCont(contLen);
if (pMeta == NULL) {
taosRUnLockLatch(&pStb->lock);
mndReleaseDb(pMnode, pDb);
mndReleaseStb(pMnode, pStb);
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("stb:%s, failed to get meta since %s", tbFName, terrstr());
return -1;
}
strcpy(pMeta->dbFName, pStb->db);
strcpy(pMeta->tbName, pInfo->tbName);
strcpy(pMeta->stbName, pInfo->tbName);
pMeta->dbId = htobe64(pDb->uid);
pMeta->numOfTags = htonl(pStb->numOfTags);
pMeta->numOfColumns = htonl(pStb->numOfColumns);
pMeta->precision = pDb->cfg.precision;
pMeta->tableType = TSDB_SUPER_TABLE;
pMeta->update = pDb->cfg.update;
pMeta->sversion = htonl(pStb->version);
pMeta->suid = htobe64(pStb->uid);
pMeta->tuid = htobe64(pStb->uid);
for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
SSchema *pSchema = &pMeta->pSchema[i];
SSchema *pSrcSchema = &pStb->pColumns[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = htonl(pSrcSchema->colId);
pSchema->bytes = htonl(pSrcSchema->bytes);
}
for (int32_t i = 0; i < pStb->numOfTags; ++i) {
SSchema *pSchema = &pMeta->pSchema[i + pStb->numOfColumns];
SSchema *pSrcSchema = &pStb->pTags[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = htonl(pSrcSchema->colId);
pSchema->bytes = htonl(pSrcSchema->bytes);
}
taosRUnLockLatch(&pStb->lock);
int32_t code = mndBuildStbSchemaImp(pDb, pStb, tbName, pRsp);
mndReleaseDb(pMnode, pDb);
mndReleaseStb(pMnode, pStb);
pReq->pCont = pMeta;
pReq->contLen = contLen;
mDebug("stb:%s, meta is retrieved, cols:%d tags:%d", tbFName, pStb->numOfColumns, pStb->numOfTags);
return 0;
return code;
}
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *stbs, int32_t num, void **rsp, int32_t *rspLen) {
SSdb *pSdb = pMnode->pSdb;
int32_t bufSize = num * (sizeof(STableMetaRsp) + 4 * sizeof(SSchema));
void *buf = malloc(bufSize);
int32_t len = 0;
int32_t contLen = 0;
STableMetaRsp *pRsp = NULL;
static int32_t mndProcessStbMetaReq(SMnodeMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
int32_t code = -1;
STableInfoReq infoReq = {0};
STableMetaRsp metaRsp = {0};
for (int32_t i = 0; i < num; ++i) {
SSTableMetaVersion *stb = &stbs[i];
stb->suid = be64toh(stb->suid);
stb->sversion = ntohs(stb->sversion);
stb->tversion = ntohs(stb->tversion);
if ((contLen + sizeof(STableMetaRsp)) > bufSize) {
bufSize = contLen + (num - i) * (sizeof(STableMetaRsp) + 4 * sizeof(SSchema));
buf = realloc(buf, bufSize);
}
pRsp = (STableMetaRsp *)((char *)buf + contLen);
strcpy(pRsp->dbFName, stb->dbFName);
strcpy(pRsp->tbName, stb->stbName);
strcpy(pRsp->stbName, stb->stbName);
mDebug("start to retrieve meta, db:%s, stb:%s", stb->dbFName, stb->stbName);
SDbObj *pDb = mndAcquireDb(pMnode, stb->dbFName);
if (pDb == NULL) {
pRsp->numOfColumns = -1;
pRsp->suid = htobe64(stb->suid);
contLen += sizeof(STableMetaRsp);
mWarn("db:%s, failed to require db since %s", stb->dbFName, terrstr());
continue;
}
char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
snprintf(tbFName, sizeof(tbFName), "%s.%s", stb->dbFName, stb->stbName);
SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
if (pStb == NULL) {
mndReleaseDb(pMnode, pDb);
pRsp->numOfColumns = -1;
pRsp->suid = htobe64(stb->suid);
contLen += sizeof(STableMetaRsp);
mWarn("stb:%s, failed to get meta since %s", tbFName, terrstr());
continue;
}
taosRLockLatch(&pStb->lock);
if (stb->suid == pStb->uid && stb->sversion == pStb->version) {
taosRUnLockLatch(&pStb->lock);
mndReleaseDb(pMnode, pDb);
mndReleaseStb(pMnode, pStb);
continue;
}
int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
int32_t len = totalCols * sizeof(SSchema);
contLen += sizeof(STableMetaRsp) + len;
if (contLen > bufSize) {
bufSize = contLen + (num - i - 1) * (sizeof(STableMetaRsp) + 4 * sizeof(SSchema));
buf = realloc(buf, bufSize);
}
pRsp->numOfTags = htonl(pStb->numOfTags);
pRsp->numOfColumns = htonl(pStb->numOfColumns);
pRsp->precision = pDb->cfg.precision;
pRsp->tableType = TSDB_SUPER_TABLE;
pRsp->update = pDb->cfg.update;
pRsp->sversion = htonl(pStb->version);
pRsp->suid = htobe64(pStb->uid);
pRsp->tuid = htobe64(pStb->uid);
for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
SSchema *pSchema = &pRsp->pSchema[i];
SSchema *pSrcSchema = &pStb->pColumns[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = htonl(pSrcSchema->colId);
pSchema->bytes = htonl(pSrcSchema->bytes);
}
for (int32_t i = 0; i < pStb->numOfTags; ++i) {
SSchema *pSchema = &pRsp->pSchema[i + pStb->numOfColumns];
SSchema *pSrcSchema = &pStb->pTags[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = htonl(pSrcSchema->colId);
pSchema->bytes = htonl(pSrcSchema->bytes);
}
taosRUnLockLatch(&pStb->lock);
mndReleaseDb(pMnode, pDb);
mndReleaseStb(pMnode, pStb);
if (tDeserializeSTableInfoReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &infoReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto RETRIEVE_META_OVER;
}
if (contLen > 0) {
*rsp = buf;
*rspLen = contLen;
} else {
*rsp = NULL;
tfree(buf);
*rspLen = 0;
mDebug("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
if (mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp) != 0) {
goto RETRIEVE_META_OVER;
}
int32_t rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto RETRIEVE_META_OVER;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto RETRIEVE_META_OVER;
}
tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp);
pReq->pCont = pRsp;
pReq->contLen = rspLen;
code = 0;
mDebug("stb:%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName);
RETRIEVE_META_OVER:
if (code != 0) {
mError("stb:%s.%s, failed to retrieve meta since %s", infoReq.dbFName, infoReq.tbName, terrstr());
}
tFreeSTableMetaRsp(&metaRsp);
return code;
}
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
int32_t *pRspLen) {
STableMetaBatchRsp batchMetaRsp = {0};
batchMetaRsp.pArray = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));
if (batchMetaRsp.pArray == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < numOfStbs; ++i) {
SSTableMetaVersion *pStbVersion = &pStbVersions[i];
pStbVersion->suid = be64toh(pStbVersion->suid);
pStbVersion->sversion = ntohs(pStbVersion->sversion);
pStbVersion->tversion = ntohs(pStbVersion->tversion);
STableMetaRsp metaRsp = {0};
mDebug("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName);
if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp) != 0) {
metaRsp.numOfColumns = -1;
metaRsp.suid = pStbVersion->suid;
}
if (pStbVersion->sversion != metaRsp.sversion) {
taosArrayPush(batchMetaRsp.pArray, &metaRsp);
}
}
int32_t rspLen = tSerializeSTableMetaBatchRsp(NULL, 0, &batchMetaRsp);
if (rspLen < 0) {
tFreeSTableMetaBatchRsp(&batchMetaRsp);
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
void *pRsp = malloc(rspLen);
if (pRsp == NULL) {
tFreeSTableMetaBatchRsp(&batchMetaRsp);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
tSerializeSTableMetaBatchRsp(pRsp, rspLen, &batchMetaRsp);
*ppRsp = pRsp;
*pRspLen = rspLen;
return 0;
}
@ -1407,33 +1375,33 @@ static int32_t mndGetStbMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
}
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "columns");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "tags");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

29
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -34,7 +34,7 @@ static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pTopic, SMqTopicObj
static int32_t mndProcessCreateTopicMsg(SMnodeMsg *pMsg);
static int32_t mndProcessDropTopicMsg(SMnodeMsg *pMsg);
static int32_t mndProcessDropTopicInRsp(SMnodeMsg *pMsg);
static int32_t mndProcessTopicMetaMsg(SMnodeMsg *pMsg);
static int32_t mndProcessTopicMetaMsg(SMnodeMsg *pReq);
static int32_t mndGetTopicMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveTopic(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter);
@ -343,11 +343,16 @@ static int32_t mndProcessDropTopicInRsp(SMnodeMsg *pMsg) {
return 0;
}
static int32_t mndProcessTopicMetaMsg(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
STableInfoReq *pInfo = pMsg->rpcMsg.pCont;
static int32_t mndProcessTopicMetaMsg(SMnodeMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
STableInfoReq infoReq = {0};
mDebug("topic:%s, start to retrieve meta", pInfo->tbName);
if (tSerializeSTableInfoReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &infoReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
mDebug("topic:%s, start to retrieve meta", infoReq.tbName);
#if 0
SDbObj *pDb = mndAcquireDbByTopic(pMnode, pInfo->tableFname);
@ -389,7 +394,7 @@ static int32_t mndProcessTopicMetaMsg(SMnodeMsg *pMsg) {
pMeta->tuid = htonl(pTopic->uid);
for (int32_t i = 0; i < totalCols; ++i) {
SSchema *pSchema = &pMeta->pSchema[i];
SSchema *pSchema = &pMeta->pSchemas[i];
SSchema *pSrcSchema = &pTopic->pSchema[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
@ -442,33 +447,33 @@ static int32_t mndGetTopicMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *
}
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "columns");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "tags");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

32
source/dnode/mnode/impl/src/mndUser.c
View File

@ -302,7 +302,10 @@ static int32_t mndProcessCreateUserReq(SMnodeMsg *pReq) {
SUserObj *pOperUser = NULL;
SCreateUserReq createReq = {0};
if (tDeserializeSCreateUserReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &createReq) != 0) goto CREATE_USER_OVER;
if (tDeserializeSCreateUserReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &createReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto CREATE_USER_OVER;
}
mDebug("user:%s, start to create", createReq.user);
@ -402,7 +405,10 @@ static int32_t mndProcessAlterUserReq(SMnodeMsg *pReq) {
SUserObj newUser = {0};
SAlterUserReq alterReq = {0};
if (tDeserializeSAlterUserReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &alterReq) != 0) goto ALTER_USER_OVER;
if (tDeserializeSAlterUserReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &alterReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto ALTER_USER_OVER;
}
mDebug("user:%s, start to alter", alterReq.user);
@ -537,7 +543,10 @@ static int32_t mndProcessDropUserReq(SMnodeMsg *pReq) {
SUserObj *pOperUser = NULL;
SDropUserReq dropReq = {0};
if (tDeserializeSDropUserReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &dropReq) != 0) goto DROP_USER_OVER;
if (tDeserializeSDropUserReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto DROP_USER_OVER;
}
mDebug("user:%s, start to drop", dropReq.user);
@ -583,7 +592,10 @@ static int32_t mndProcessGetUserAuthReq(SMnodeMsg *pReq) {
SGetUserAuthReq authReq = {0};
SGetUserAuthRsp authRsp = {0};
if (tDeserializeSGetUserAuthReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &authReq) != 0) goto GET_AUTH_OVER;
if (tDeserializeSGetUserAuthReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &authReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto GET_AUTH_OVER;
}
mTrace("user:%s, start to get auth", authReq.user);
@ -640,33 +652,33 @@ static int32_t mndGetUserMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *p
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_USER_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 10 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "privilege");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_USER_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "account");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

20
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -488,35 +488,35 @@ static int32_t mndGetVgroupMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp
}
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "vgId");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "tables");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
for (int32_t i = 0; i < pShow->replica; ++i) {
pShow->bytes[cols] = 2;
pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
snprintf(pSchema[cols].name, TSDB_COL_NAME_LEN, "v%d_dnode", i + 1);
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 9 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
snprintf(pSchema[cols].name, TSDB_COL_NAME_LEN, "v%d_status", i + 1);
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
}
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
@ -608,21 +608,21 @@ static int32_t mndGetVnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchema;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "vgId");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 12 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "status");
pSchema[cols].bytes = htonl(pShow->bytes[cols]);
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = htonl(cols);
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;

64
source/dnode/mnode/impl/test/stb/stb.cpp
View File

@ -339,45 +339,37 @@ TEST_F(MndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
// ----- meta ------
{
int32_t contLen = sizeof(STableInfoReq);
STableInfoReq* pReq = (STableInfoReq*)rpcMallocCont(contLen);
strcpy(pReq->dbFName, dbname);
strcpy(pReq->tbName, "stb");
STableInfoReq infoReq = {0};
strcpy(infoReq.dbFName, dbname);
strcpy(infoReq.tbName, "stb");
int32_t contLen = tSerializeSTableInfoReq(NULL, 0, &infoReq);
void* pReq = rpcMallocCont(contLen);
tSerializeSTableInfoReq(pReq, contLen, &infoReq);
SRpcMsg* pMsg = test.SendReq(TDMT_MND_STB_META, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
STableMetaRsp* pRsp = (STableMetaRsp*)pMsg->pCont;
pRsp->numOfTags = htonl(pRsp->numOfTags);
pRsp->numOfColumns = htonl(pRsp->numOfColumns);
pRsp->sversion = htonl(pRsp->sversion);
pRsp->tversion = htonl(pRsp->tversion);
pRsp->suid = be64toh(pRsp->suid);
pRsp->tuid = be64toh(pRsp->tuid);
pRsp->vgId = be64toh(pRsp->vgId);
for (int32_t i = 0; i < pRsp->numOfTags + pRsp->numOfColumns; ++i) {
SSchema* pSchema = &pRsp->pSchema[i];
pSchema->colId = htonl(pSchema->colId);
pSchema->bytes = htonl(pSchema->bytes);
}
STableMetaRsp metaRsp = {0};
tDeserializeSTableMetaRsp(pMsg->pCont, pMsg->contLen, &metaRsp);
EXPECT_STREQ(pRsp->dbFName, dbname);
EXPECT_STREQ(pRsp->tbName, "stb");
EXPECT_STREQ(pRsp->stbName, "stb");
EXPECT_EQ(pRsp->numOfColumns, 2);
EXPECT_EQ(pRsp->numOfTags, 3);
EXPECT_EQ(pRsp->precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(pRsp->tableType, TSDB_SUPER_TABLE);
EXPECT_EQ(pRsp->update, 0);
EXPECT_EQ(pRsp->sversion, 1);
EXPECT_EQ(pRsp->tversion, 0);
EXPECT_GT(pRsp->suid, 0);
EXPECT_GT(pRsp->tuid, 0);
EXPECT_EQ(pRsp->vgId, 0);
EXPECT_STREQ(metaRsp.dbFName, dbname);
EXPECT_STREQ(metaRsp.tbName, "stb");
EXPECT_STREQ(metaRsp.stbName, "stb");
EXPECT_EQ(metaRsp.numOfColumns, 2);
EXPECT_EQ(metaRsp.numOfTags, 3);
EXPECT_EQ(metaRsp.precision, TSDB_TIME_PRECISION_MILLI);
EXPECT_EQ(metaRsp.tableType, TSDB_SUPER_TABLE);
EXPECT_EQ(metaRsp.update, 0);
EXPECT_EQ(metaRsp.sversion, 1);
EXPECT_EQ(metaRsp.tversion, 0);
EXPECT_GT(metaRsp.suid, 0);
EXPECT_GT(metaRsp.tuid, 0);
EXPECT_EQ(metaRsp.vgId, 0);
{
SSchema* pSchema = &pRsp->pSchema[0];
SSchema* pSchema = &metaRsp.pSchemas[0];
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_TIMESTAMP);
EXPECT_EQ(pSchema->colId, 1);
EXPECT_EQ(pSchema->bytes, 8);
@ -385,7 +377,7 @@ TEST_F(MndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
}
{
SSchema* pSchema = &pRsp->pSchema[1];
SSchema* pSchema = &metaRsp.pSchemas[1];
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->colId, 2);
EXPECT_EQ(pSchema->bytes, 12);
@ -393,7 +385,7 @@ TEST_F(MndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
}
{
SSchema* pSchema = &pRsp->pSchema[2];
SSchema* pSchema = &metaRsp.pSchemas[2];
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_TINYINT);
EXPECT_EQ(pSchema->colId, 3);
EXPECT_EQ(pSchema->bytes, 2);
@ -401,7 +393,7 @@ TEST_F(MndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
}
{
SSchema* pSchema = &pRsp->pSchema[3];
SSchema* pSchema = &metaRsp.pSchemas[3];
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BIGINT);
EXPECT_EQ(pSchema->colId, 4);
EXPECT_EQ(pSchema->bytes, 8);
@ -409,12 +401,14 @@ TEST_F(MndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
}
{
SSchema* pSchema = &pRsp->pSchema[4];
SSchema* pSchema = &metaRsp.pSchemas[4];
EXPECT_EQ(pSchema->type, TSDB_DATA_TYPE_BINARY);
EXPECT_EQ(pSchema->colId, 5);
EXPECT_EQ(pSchema->bytes, 16);
EXPECT_STREQ(pSchema->name, "tag3");
}
tFreeSTableMetaRsp(&metaRsp);
}
// restart

67
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -71,7 +71,6 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) {
}
static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
STableInfoReq * pReq = (STableInfoReq *)(pMsg->pCont);
STbCfg * pTbCfg = NULL;
STbCfg * pStbCfg = NULL;
tb_uid_t uid;
@ -79,12 +78,19 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
int32_t nTagCols;
SSchemaWrapper *pSW = NULL;
STableMetaRsp *pTbMetaMsg = NULL;
SSchema * pTagSchema;
STableMetaRsp metaRsp = {0};
SSchema *pTagSchema;
SRpcMsg rpcMsg;
int msgLen = 0;
int32_t code = TSDB_CODE_VND_APP_ERROR;
pTbCfg = metaGetTbInfoByName(pVnode->pMeta, pReq->tbName, &uid);
STableInfoReq infoReq = {0};
if (tDeserializeSTableInfoReq(pMsg->pCont, pMsg->contLen, &infoReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto _exit;
}
pTbCfg = metaGetTbInfoByName(pVnode->pMeta, infoReq.tbName, &uid);
if (pTbCfg == NULL) {
code = TSDB_CODE_VND_TB_NOT_EXIST;
goto _exit;
@ -114,44 +120,51 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
pTagSchema = NULL;
}
msgLen = sizeof(STableMetaRsp) + sizeof(SSchema) * (nCols + nTagCols);
pTbMetaMsg = (STableMetaRsp *)rpcMallocCont(msgLen);
if (pTbMetaMsg == NULL) {
metaRsp.pSchemas = calloc(nCols + nTagCols, sizeof(SSchema));
if (metaRsp.pSchemas == NULL) {
code = TSDB_CODE_VND_OUT_OF_MEMORY;
goto _exit;
}
pTbMetaMsg->dbId = htobe64(pVnode->config.dbId);
memcpy(pTbMetaMsg->dbFName, pReq->dbFName, sizeof(pTbMetaMsg->dbFName));
strcpy(pTbMetaMsg->tbName, pReq->tbName);
metaRsp.dbId = htobe64(pVnode->config.dbId);
memcpy(metaRsp.dbFName, infoReq.dbFName, sizeof(metaRsp.dbFName));
strcpy(metaRsp.tbName, infoReq.tbName);
if (pTbCfg->type == META_CHILD_TABLE) {
strcpy(pTbMetaMsg->stbName, pStbCfg->name);
pTbMetaMsg->suid = htobe64(pTbCfg->ctbCfg.suid);
strcpy(metaRsp.stbName, pStbCfg->name);
metaRsp.suid = pTbCfg->ctbCfg.suid;
} else if (pTbCfg->type == META_SUPER_TABLE) {
strcpy(pTbMetaMsg->stbName, pTbCfg->name);
pTbMetaMsg->suid = htobe64(uid);
strcpy(metaRsp.stbName, pTbCfg->name);
metaRsp.suid = uid;
}
pTbMetaMsg->numOfTags = htonl(nTagCols);
pTbMetaMsg->numOfColumns = htonl(nCols);
pTbMetaMsg->tableType = pTbCfg->type;
pTbMetaMsg->tuid = htobe64(uid);
pTbMetaMsg->vgId = htonl(pVnode->vgId);
metaRsp.numOfTags = nTagCols;
metaRsp.numOfColumns = nCols;
metaRsp.tableType = pTbCfg->type;
metaRsp.tuid = uid;
metaRsp.vgId = pVnode->vgId;
memcpy(pTbMetaMsg->pSchema, pSW->pSchema, sizeof(SSchema) * pSW->nCols);
memcpy(metaRsp.pSchemas, pSW->pSchema, sizeof(SSchema) * pSW->nCols);
if (nTagCols) {
memcpy(POINTER_SHIFT(pTbMetaMsg->pSchema, sizeof(SSchema) * pSW->nCols), pTagSchema, sizeof(SSchema) * nTagCols);
memcpy(POINTER_SHIFT(metaRsp.pSchemas, sizeof(SSchema) * pSW->nCols), pTagSchema, sizeof(SSchema) * nTagCols);
}
for (int i = 0; i < nCols + nTagCols; i++) {
SSchema *pSch = pTbMetaMsg->pSchema + i;
pSch->colId = htonl(pSch->colId);
pSch->bytes = htonl(pSch->bytes);
int32_t rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
if (rspLen < 0) {
code = TSDB_CODE_INVALID_MSG;
goto _exit;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp);
code = 0;
_exit:
tFreeSTableMetaRsp(&metaRsp);
if (pSW != NULL) {
tfree(pSW->pSchema);
tfree(pSW);
@ -170,13 +183,13 @@ _exit:
rpcMsg.handle = pMsg->handle;
rpcMsg.ahandle = pMsg->ahandle;
rpcMsg.pCont = pTbMetaMsg;
rpcMsg.contLen = msgLen;
rpcMsg.pCont = pRsp;
rpcMsg.contLen = rspLen;
rpcMsg.code = code;
rpcSendResponse(&rpcMsg);
return 0;
return code;
}
static void freeItemHelper(void *pItem) {

249
source/libs/catalog/test/catalogTests.cpp
View File

@ -222,7 +222,7 @@ void ctgTestBuildDBVgroup(SDBVgInfo **pdbVgroup) {
for (int32_t n = 0; n < vgInfo.epset.numOfEps; ++n) {
SEp *addr = &vgInfo.epset.eps[n];
strcpy(addr->fqdn, "a0");
addr->port = htons(n + 22);
addr->port = n + 22;
}
taosHashPut(dbVgroup->vgHash, &vgInfo.vgId, sizeof(vgInfo.vgId), &vgInfo, sizeof(vgInfo));
@ -247,19 +247,19 @@ void ctgTestBuildSTableMetaRsp(STableMetaRsp *rspMsg) {
rspMsg->vgId = 1;
SSchema *s = NULL;
s = &rspMsg->pSchema[0];
s = &rspMsg->pSchemas[0];
s->type = TSDB_DATA_TYPE_TIMESTAMP;
s->colId = 1;
s->bytes = 8;
strcpy(s->name, "ts");
s = &rspMsg->pSchema[1];
s = &rspMsg->pSchemas[1];
s->type = TSDB_DATA_TYPE_INT;
s->colId = 2;
s->bytes = 4;
strcpy(s->name, "col1s");
s = &rspMsg->pSchema[2];
s = &rspMsg->pSchemas[2];
s->type = TSDB_DATA_TYPE_BINARY;
s->colId = 3;
s->bytes = 12 + 1;
@ -309,173 +309,189 @@ void ctgTestRspDbVgroups(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *
}
void ctgTestRspTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
pRsp->code = 0;
pRsp->contLen = sizeof(STableMetaRsp) + (ctgTestColNum + ctgTestTagNum) * sizeof(SSchema);
pRsp->pCont = calloc(1, pRsp->contLen);
rspMsg = (STableMetaRsp *)pRsp->pCont;
strcpy(rspMsg->dbFName, ctgTestDbname);
strcpy(rspMsg->tbName, ctgTestTablename);
rspMsg->numOfTags = 0;
rspMsg->numOfColumns = htonl(ctgTestColNum);
rspMsg->precision = 1;
rspMsg->tableType = TSDB_NORMAL_TABLE;
rspMsg->update = 1;
rspMsg->sversion = htonl(ctgTestSVersion);
rspMsg->tversion = htonl(ctgTestTVersion);
rspMsg->suid = 0;
rspMsg->tuid = htobe64(0x0000000000000001);
rspMsg->vgId = htonl(8);
STableMetaRsp metaRsp = {0};
strcpy(metaRsp.dbFName, ctgTestDbname);
strcpy(metaRsp.tbName, ctgTestTablename);
metaRsp.numOfTags = 0;
metaRsp.numOfColumns = ctgTestColNum;
metaRsp.precision = 1;
metaRsp.tableType = TSDB_NORMAL_TABLE;
metaRsp.update = 1;
metaRsp.sversion = ctgTestSVersion;
metaRsp.tversion = ctgTestTVersion;
metaRsp.suid = 0;
metaRsp.tuid = 0x0000000000000001;
metaRsp.vgId = 8;
metaRsp.pSchemas = (SSchema *)malloc((metaRsp.numOfTags + metaRsp.numOfColumns) * sizeof(SSchema));
SSchema *s = NULL;
s = &rspMsg->pSchema[0];
s = &metaRsp.pSchemas[0];
s->type = TSDB_DATA_TYPE_TIMESTAMP;
s->colId = htonl(1);
s->bytes = htonl(8);
s->colId = 1;
s->bytes = 8;
strcpy(s->name, "ts");
s = &rspMsg->pSchema[1];
s = &metaRsp.pSchemas[1];
s->type = TSDB_DATA_TYPE_INT;
s->colId = htonl(2);
s->bytes = htonl(4);
s->colId = 2;
s->bytes = 4;
strcpy(s->name, "col1");
return;
int32_t contLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
void *pReq = rpcMallocCont(contLen);
tSerializeSTableMetaRsp(pReq, contLen, &metaRsp);
pRsp->code = 0;
pRsp->contLen = contLen;
pRsp->pCont = pReq;
tFreeSTableMetaRsp(&metaRsp);
}
void ctgTestRspCTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
pRsp->code = 0;
pRsp->contLen = sizeof(STableMetaRsp) + (ctgTestColNum + ctgTestTagNum) * sizeof(SSchema);
pRsp->pCont = calloc(1, pRsp->contLen);
rspMsg = (STableMetaRsp *)pRsp->pCont;
strcpy(rspMsg->dbFName, ctgTestDbname);
strcpy(rspMsg->tbName, ctgTestCTablename);
strcpy(rspMsg->stbName, ctgTestSTablename);
rspMsg->numOfTags = htonl(ctgTestTagNum);
rspMsg->numOfColumns = htonl(ctgTestColNum);
rspMsg->precision = 1;
rspMsg->tableType = TSDB_CHILD_TABLE;
rspMsg->update = 1;
rspMsg->sversion = htonl(ctgTestSVersion);
rspMsg->tversion = htonl(ctgTestTVersion);
rspMsg->suid = htobe64(0x0000000000000002);
rspMsg->tuid = htobe64(0x0000000000000003);
rspMsg->vgId = htonl(9);
STableMetaRsp metaRsp = {0};
strcpy(metaRsp.dbFName, ctgTestDbname);
strcpy(metaRsp.tbName, ctgTestCTablename);
strcpy(metaRsp.stbName, ctgTestSTablename);
metaRsp.numOfTags = ctgTestTagNum;
metaRsp.numOfColumns = ctgTestColNum;
metaRsp.precision = 1;
metaRsp.tableType = TSDB_CHILD_TABLE;
metaRsp.update = 1;
metaRsp.sversion = ctgTestSVersion;
metaRsp.tversion = ctgTestTVersion;
metaRsp.suid = 0x0000000000000002;
metaRsp.tuid = 0x0000000000000003;
metaRsp.vgId = 9;
metaRsp.pSchemas = (SSchema *)malloc((metaRsp.numOfTags + metaRsp.numOfColumns) * sizeof(SSchema));
SSchema *s = NULL;
s = &rspMsg->pSchema[0];
s = &metaRsp.pSchemas[0];
s->type = TSDB_DATA_TYPE_TIMESTAMP;
s->colId = htonl(1);
s->bytes = htonl(8);
s->colId = 1;
s->bytes = 8;
strcpy(s->name, "ts");
s = &rspMsg->pSchema[1];
s = &metaRsp.pSchemas[1];
s->type = TSDB_DATA_TYPE_INT;
s->colId = htonl(2);
s->bytes = htonl(4);
s->colId = 2;
s->bytes = 4;
strcpy(s->name, "col1s");
s = &rspMsg->pSchema[2];
s = &metaRsp.pSchemas[2];
s->type = TSDB_DATA_TYPE_BINARY;
s->colId = htonl(3);
s->bytes = htonl(12);
s->colId = 3;
s->bytes = 12;
strcpy(s->name, "tag1s");
return;
int32_t contLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
void *pReq = rpcMallocCont(contLen);
tSerializeSTableMetaRsp(pReq, contLen, &metaRsp);
pRsp->code = 0;
pRsp->contLen = contLen;
pRsp->pCont = pReq;
tFreeSTableMetaRsp(&metaRsp);
}
void ctgTestRspSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
pRsp->code = 0;
pRsp->contLen = sizeof(STableMetaRsp) + (ctgTestColNum + ctgTestTagNum) * sizeof(SSchema);
pRsp->pCont = calloc(1, pRsp->contLen);
rspMsg = (STableMetaRsp *)pRsp->pCont;
strcpy(rspMsg->dbFName, ctgTestDbname);
strcpy(rspMsg->tbName, ctgTestSTablename);
strcpy(rspMsg->stbName, ctgTestSTablename);
rspMsg->numOfTags = htonl(ctgTestTagNum);
rspMsg->numOfColumns = htonl(ctgTestColNum);
rspMsg->precision = 1;
rspMsg->tableType = TSDB_SUPER_TABLE;
rspMsg->update = 1;
rspMsg->sversion = htonl(ctgTestSVersion);
rspMsg->tversion = htonl(ctgTestTVersion);
rspMsg->suid = htobe64(ctgTestSuid);
rspMsg->tuid = htobe64(ctgTestSuid);
rspMsg->vgId = 0;
STableMetaRsp metaRsp = {0};
strcpy(metaRsp.dbFName, ctgTestDbname);
strcpy(metaRsp.tbName, ctgTestSTablename);
strcpy(metaRsp.stbName, ctgTestSTablename);
metaRsp.numOfTags = ctgTestTagNum;
metaRsp.numOfColumns = ctgTestColNum;
metaRsp.precision = 1;
metaRsp.tableType = TSDB_SUPER_TABLE;
metaRsp.update = 1;
metaRsp.sversion = ctgTestSVersion;
metaRsp.tversion = ctgTestTVersion;
metaRsp.suid = ctgTestSuid;
metaRsp.tuid = ctgTestSuid;
metaRsp.vgId = 0;
metaRsp.pSchemas = (SSchema *)malloc((metaRsp.numOfTags + metaRsp.numOfColumns) * sizeof(SSchema));
SSchema *s = NULL;
s = &rspMsg->pSchema[0];
s = &metaRsp.pSchemas[0];
s->type = TSDB_DATA_TYPE_TIMESTAMP;
s->colId = htonl(1);
s->bytes = htonl(8);
s->colId = 1;
s->bytes = 8;
strcpy(s->name, "ts");
s = &rspMsg->pSchema[1];
s = &metaRsp.pSchemas[1];
s->type = TSDB_DATA_TYPE_INT;
s->colId = htonl(2);
s->bytes = htonl(4);
s->colId = 2;
s->bytes = 4;
strcpy(s->name, "col1s");
s = &rspMsg->pSchema[2];
s = &metaRsp.pSchemas[2];
s->type = TSDB_DATA_TYPE_BINARY;
s->colId = htonl(3);
s->bytes = htonl(12);
s->colId = 3;
s->bytes = 12;
strcpy(s->name, "tag1s");
return;
int32_t contLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
void *pReq = rpcMallocCont(contLen);
tSerializeSTableMetaRsp(pReq, contLen, &metaRsp);
pRsp->code = 0;
pRsp->contLen = contLen;
pRsp->pCont = pReq;
tFreeSTableMetaRsp(&metaRsp);
}
void ctgTestRspMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
static int32_t idx = 1;
pRsp->code = 0;
pRsp->contLen = sizeof(STableMetaRsp) + (ctgTestColNum + ctgTestTagNum) * sizeof(SSchema);
pRsp->pCont = calloc(1, pRsp->contLen);
rspMsg = (STableMetaRsp *)pRsp->pCont;
strcpy(rspMsg->dbFName, ctgTestDbname);
sprintf(rspMsg->tbName, "%s_%d", ctgTestSTablename, idx);
sprintf(rspMsg->stbName, "%s_%d", ctgTestSTablename, idx);
rspMsg->numOfTags = htonl(ctgTestTagNum);
rspMsg->numOfColumns = htonl(ctgTestColNum);
rspMsg->precision = 1;
rspMsg->tableType = TSDB_SUPER_TABLE;
rspMsg->update = 1;
rspMsg->sversion = htonl(ctgTestSVersion);
rspMsg->tversion = htonl(ctgTestTVersion);
rspMsg->suid = htobe64(ctgTestSuid + idx);
rspMsg->tuid = htobe64(ctgTestSuid + idx);
rspMsg->vgId = 0;
STableMetaRsp metaRsp = {0};
strcpy(metaRsp.dbFName, ctgTestDbname);
sprintf(metaRsp.tbName, "%s_%d", ctgTestSTablename, idx);
sprintf(metaRsp.stbName, "%s_%d", ctgTestSTablename, idx);
metaRsp.numOfTags = ctgTestTagNum;
metaRsp.numOfColumns = ctgTestColNum;
metaRsp.precision = 1;
metaRsp.tableType = TSDB_SUPER_TABLE;
metaRsp.update = 1;
metaRsp.sversion = ctgTestSVersion;
metaRsp.tversion = ctgTestTVersion;
metaRsp.suid = ctgTestSuid + idx;
metaRsp.tuid = ctgTestSuid + idx;
metaRsp.vgId = 0;
metaRsp.pSchemas = (SSchema *)malloc((metaRsp.numOfTags + metaRsp.numOfColumns) * sizeof(SSchema));
SSchema *s = NULL;
s = &rspMsg->pSchema[0];
s = &metaRsp.pSchemas[0];
s->type = TSDB_DATA_TYPE_TIMESTAMP;
s->colId = htonl(1);
s->bytes = htonl(8);
s->colId = 1;
s->bytes = 8;
strcpy(s->name, "ts");
s = &rspMsg->pSchema[1];
s = &metaRsp.pSchemas[1];
s->type = TSDB_DATA_TYPE_INT;
s->colId = htonl(2);
s->bytes = htonl(4);
s->colId = 2;
s->bytes = 4;
strcpy(s->name, "col1s");
s = &rspMsg->pSchema[2];
s = &metaRsp.pSchemas[2];
s->type = TSDB_DATA_TYPE_BINARY;
s->colId = htonl(3);
s->bytes = htonl(12);
s->colId = 3;
s->bytes = 12;
strcpy(s->name, "tag1s");
++idx;
return;
}
int32_t contLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
void *pReq = rpcMallocCont(contLen);
tSerializeSTableMetaRsp(pReq, contLen, &metaRsp);
pRsp->code = 0;
pRsp->contLen = contLen;
pRsp->pCont = pReq;
tFreeSTableMetaRsp(&metaRsp);
}
void ctgTestRspByIdx(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
switch (ctgTestRspFunc[ctgTestRspIdx]) {
@ -503,7 +519,6 @@ void ctgTestRspByIdx(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp
return;
}
void ctgTestRspDbVgroupsAndNormalMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestRspDbVgroups(shandle, pEpSet, pMsg, pRsp);

203
source/libs/qcom/src/querymsg.c
View File

@ -21,51 +21,42 @@
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wformat-truncation"
int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen) = {0};
int32_t (*queryBuildMsg[TDMT_MAX])(void *input, char **msg, int32_t msgSize, int32_t *msgLen) = {0};
int32_t (*queryProcessMsgRsp[TDMT_MAX])(void *output, char *msg, int32_t msgSize) = {0};
int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char *msg, int32_t msgSize) = {0};
int32_t queryBuildTableMetaReqMsg(void* input, char **msg, int32_t msgSize, int32_t *msgLen) {
int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) {
SBuildTableMetaInput *pInput = input;
if (NULL == input || NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT;
}
SBuildTableMetaInput* bInput = (SBuildTableMetaInput *)input;
int32_t estimateSize = sizeof(STableInfoReq);
if (NULL == *msg || msgSize < estimateSize) {
tfree(*msg);
*msg = rpcMallocCont(estimateSize);
if (NULL == *msg) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
STableInfoReq infoReq = {0};
infoReq.header.vgId = pInput->vgId;
if (pInput->dbFName) {
tstrncpy(infoReq.dbFName, pInput->dbFName, TSDB_DB_FNAME_LEN);
}
tstrncpy(infoReq.tbName, pInput->tbName, TSDB_TABLE_NAME_LEN);
STableInfoReq *bMsg = (STableInfoReq *)*msg;
int32_t bufLen = tSerializeSTableInfoReq(NULL, 0, &infoReq);
void *pBuf = rpcMallocCont(bufLen);
tSerializeSTableInfoReq(pBuf, bufLen, &infoReq);
bMsg->header.vgId = htonl(bInput->vgId);
*msg = pBuf;
*msgLen = bufLen;
if (bInput->dbFName) {
tstrncpy(bMsg->dbFName, bInput->dbFName, tListLen(bMsg->dbFName));
}
tstrncpy(bMsg->tbName, bInput->tbName, tListLen(bMsg->tbName));
*msgLen = (int32_t)sizeof(*bMsg);
return TSDB_CODE_SUCCESS;
}
int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) {
if (NULL == input || NULL == msg || NULL == msgLen) {
SBuildUseDBInput *pInput = input;
if (NULL == pInput || NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT;
}
SBuildUseDBInput *bInput = input;
SUseDbReq usedbReq = {0};
strncpy(usedbReq.db, bInput->db, sizeof(usedbReq.db));
strncpy(usedbReq.db, pInput->db, sizeof(usedbReq.db));
usedbReq.db[sizeof(usedbReq.db) - 1] = 0;
usedbReq.vgVersion = bInput->vgVersion;
usedbReq.vgVersion = pInput->vgVersion;
int32_t bufLen = tSerializeSUseDbReq(NULL, 0, &usedbReq);
void *pBuf = rpcMallocCont(bufLen);
@ -78,75 +69,69 @@ int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *ms
}
int32_t queryProcessUseDBRsp(void *output, char *msg, int32_t msgSize) {
SUseDbOutput *pOut = output;
SUseDbRsp usedbRsp = {0};
int32_t code = -1;
if (NULL == output || NULL == msg || msgSize <= 0) {
return TSDB_CODE_TSC_INVALID_INPUT;
code = TSDB_CODE_TSC_INVALID_INPUT;
goto PROCESS_USEDB_OVER;
}
SUseDbOutput *pOut = (SUseDbOutput *)output;
int32_t code = 0;
SUseDbRsp usedbRsp = {0};
if (tDeserializeSUseDbRsp(msg, msgSize, &usedbRsp) != 0) {
qError("invalid use db rsp msg, msgSize:%d", msgSize);
return TSDB_CODE_INVALID_MSG;
code = TSDB_CODE_INVALID_MSG;
goto PROCESS_USEDB_OVER;
}
if (usedbRsp.vgNum < 0) {
qError("invalid db[%s] vgroup number[%d]", usedbRsp.db, usedbRsp.vgNum);
return TSDB_CODE_TSC_INVALID_VALUE;
code = TSDB_CODE_TSC_INVALID_VALUE;
goto PROCESS_USEDB_OVER;
}
memcpy(pOut->db, usedbRsp.db, TSDB_DB_FNAME_LEN);
pOut->dbId = usedbRsp.uid;
pOut->dbVgroup = calloc(1, sizeof(SDBVgInfo));
if (NULL == pOut->dbVgroup) {
qError("calloc %d failed", (int32_t)sizeof(SDBVgInfo));
return TSDB_CODE_TSC_OUT_OF_MEMORY;
code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto PROCESS_USEDB_OVER;
}
pOut->dbId = usedbRsp.uid;
pOut->dbVgroup->vgVersion = usedbRsp.vgVersion;
pOut->dbVgroup->hashMethod = usedbRsp.hashMethod;
pOut->dbVgroup->vgHash =
taosHashInit(usedbRsp.vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
if (NULL == pOut->dbVgroup->vgHash) {
qError("taosHashInit %d failed", usedbRsp.vgNum);
tfree(pOut->dbVgroup);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto PROCESS_USEDB_OVER;
}
for (int32_t i = 0; i < usedbRsp.vgNum; ++i) {
SVgroupInfo *pVgInfo = taosArrayGet(usedbRsp.pVgroupInfos, i);
if (0 != taosHashPut(pOut->dbVgroup->vgHash, &pVgInfo->vgId, sizeof(int32_t), pVgInfo, sizeof(SVgroupInfo))) {
qError("taosHashPut failed");
goto _return;
code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto PROCESS_USEDB_OVER;
}
}
memcpy(pOut->db, usedbRsp.db, TSDB_DB_FNAME_LEN);
code = 0;
return code;
_return:
tFreeSUsedbRsp(&usedbRsp);
if (pOut) {
taosHashCleanup(pOut->dbVgroup->vgHash);
tfree(pOut->dbVgroup);
PROCESS_USEDB_OVER:
if (code != 0) {
if (pOut) {
if (pOut->dbVgroup) taosHashCleanup(pOut->dbVgroup->vgHash);
tfree(pOut->dbVgroup);
}
qError("failed to process usedb rsp since %s", terrstr());
}
tFreeSUsedbRsp(&usedbRsp);
return code;
}
static int32_t queryConvertTableMetaMsg(STableMetaRsp* pMetaMsg) {
pMetaMsg->dbId = be64toh(pMetaMsg->dbId);
pMetaMsg->numOfTags = ntohl(pMetaMsg->numOfTags);
pMetaMsg->numOfColumns = ntohl(pMetaMsg->numOfColumns);
pMetaMsg->sversion = ntohl(pMetaMsg->sversion);
pMetaMsg->tversion = ntohl(pMetaMsg->tversion);
pMetaMsg->tuid = be64toh(pMetaMsg->tuid);
pMetaMsg->suid = be64toh(pMetaMsg->suid);
pMetaMsg->vgId = ntohl(pMetaMsg->vgId);
static int32_t queryConvertTableMetaMsg(STableMetaRsp *pMetaMsg) {
if (pMetaMsg->numOfTags < 0 || pMetaMsg->numOfTags > TSDB_MAX_TAGS) {
qError("invalid numOfTags[%d] in table meta rsp msg", pMetaMsg->numOfTags);
return TSDB_CODE_TSC_INVALID_VALUE;
@ -157,7 +142,8 @@ static int32_t queryConvertTableMetaMsg(STableMetaRsp* pMetaMsg) {
return TSDB_CODE_TSC_INVALID_VALUE;
}
if (pMetaMsg->tableType != TSDB_SUPER_TABLE && pMetaMsg->tableType != TSDB_CHILD_TABLE && pMetaMsg->tableType != TSDB_NORMAL_TABLE) {
if (pMetaMsg->tableType != TSDB_SUPER_TABLE && pMetaMsg->tableType != TSDB_CHILD_TABLE &&
pMetaMsg->tableType != TSDB_NORMAL_TABLE) {
qError("invalid tableType[%d] in table meta rsp msg", pMetaMsg->tableType);
return TSDB_CODE_TSC_INVALID_VALUE;
}
@ -171,30 +157,20 @@ static int32_t queryConvertTableMetaMsg(STableMetaRsp* pMetaMsg) {
qError("invalid tversion[%d] in table meta rsp msg", pMetaMsg->tversion);
return TSDB_CODE_TSC_INVALID_VALUE;
}
SSchema* pSchema = pMetaMsg->pSchema;
int32_t numOfTotalCols = pMetaMsg->numOfColumns + pMetaMsg->numOfTags;
for (int i = 0; i < numOfTotalCols; ++i) {
pSchema->bytes = ntohl(pSchema->bytes);
pSchema->colId = ntohl(pSchema->colId);
pSchema++;
}
if (pMetaMsg->pSchema[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
qError("invalid colId[%d] for the first column in table meta rsp msg", pMetaMsg->pSchema[0].colId);
if (pMetaMsg->pSchemas[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
qError("invalid colId[%d] for the first column in table meta rsp msg", pMetaMsg->pSchemas[0].colId);
return TSDB_CODE_TSC_INVALID_VALUE;
}
return TSDB_CODE_SUCCESS;
}
int32_t queryCreateTableMetaFromMsg(STableMetaRsp* msg, bool isSuperTable, STableMeta **pMeta) {
int32_t queryCreateTableMetaFromMsg(STableMetaRsp *msg, bool isSuperTable, STableMeta **pMeta) {
int32_t total = msg->numOfColumns + msg->numOfTags;
int32_t metaSize = sizeof(STableMeta) + sizeof(SSchema) * total;
STableMeta* pTableMeta = calloc(1, metaSize);
STableMeta *pTableMeta = calloc(1, metaSize);
if (NULL == pTableMeta) {
qError("calloc size[%d] failed", metaSize);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
@ -202,7 +178,7 @@ int32_t queryCreateTableMetaFromMsg(STableMetaRsp* msg, bool isSuperTable, STabl
pTableMeta->vgId = isSuperTable ? 0 : msg->vgId;
pTableMeta->tableType = isSuperTable ? TSDB_SUPER_TABLE : msg->tableType;
pTableMeta->uid = isSuperTable ? msg->suid : msg->tuid;
pTableMeta->uid = isSuperTable ? msg->suid : msg->tuid;
pTableMeta->suid = msg->suid;
pTableMeta->sversion = msg->sversion;
pTableMeta->tversion = msg->tversion;
@ -211,60 +187,71 @@ int32_t queryCreateTableMetaFromMsg(STableMetaRsp* msg, bool isSuperTable, STabl
pTableMeta->tableInfo.precision = msg->precision;
pTableMeta->tableInfo.numOfColumns = msg->numOfColumns;
memcpy(pTableMeta->schema, msg->pSchema, sizeof(SSchema) * total);
memcpy(pTableMeta->schema, msg->pSchemas, sizeof(SSchema) * total);
for(int32_t i = 0; i < msg->numOfColumns; ++i) {
for (int32_t i = 0; i < msg->numOfColumns; ++i) {
pTableMeta->tableInfo.rowSize += pTableMeta->schema[i].bytes;
}
*pMeta = pTableMeta;
return TSDB_CODE_SUCCESS;
}
int32_t queryProcessTableMetaRsp(void *output, char *msg, int32_t msgSize) {
int32_t code = -1;
STableMetaRsp metaRsp = {0};
int32_t queryProcessTableMetaRsp(void* output, char *msg, int32_t msgSize) {
STableMetaRsp *pMetaMsg = (STableMetaRsp *)msg;
int32_t code = queryConvertTableMetaMsg(pMetaMsg);
if (NULL == output || NULL == msg || msgSize <= 0) {
code = TSDB_CODE_TSC_INVALID_INPUT;
goto PROCESS_META_OVER;
}
if (tDeserializeSTableMetaRsp(msg, msgSize, &metaRsp) != 0) {
code = TSDB_CODE_INVALID_MSG;
goto PROCESS_META_OVER;
}
code = queryConvertTableMetaMsg(&metaRsp);
if (code != TSDB_CODE_SUCCESS) {
return code;
goto PROCESS_META_OVER;
}
STableMetaOutput *pOut = (STableMetaOutput *)output;
if (!tIsValidSchema(pMetaMsg->pSchema, pMetaMsg->numOfColumns, pMetaMsg->numOfTags)) {
qError("validate table meta schema in rsp msg failed");
return TSDB_CODE_TSC_INVALID_VALUE;
if (!tIsValidSchema(metaRsp.pSchemas, metaRsp.numOfColumns, metaRsp.numOfTags)) {
code = TSDB_CODE_TSC_INVALID_VALUE;
goto PROCESS_META_OVER;
}
strcpy(pOut->dbFName, pMetaMsg->dbFName);
pOut->dbId = pMetaMsg->dbId;
STableMetaOutput *pOut = output;
strcpy(pOut->dbFName, metaRsp.dbFName);
pOut->dbId = metaRsp.dbId;
if (pMetaMsg->tableType == TSDB_CHILD_TABLE) {
if (metaRsp.tableType == TSDB_CHILD_TABLE) {
SET_META_TYPE_BOTH_TABLE(pOut->metaType);
strcpy(pOut->ctbName, pMetaMsg->tbName);
strcpy(pOut->tbName, pMetaMsg->stbName);
pOut->ctbMeta.vgId = pMetaMsg->vgId;
pOut->ctbMeta.tableType = pMetaMsg->tableType;
pOut->ctbMeta.uid = pMetaMsg->tuid;
pOut->ctbMeta.suid = pMetaMsg->suid;
strcpy(pOut->ctbName, metaRsp.tbName);
strcpy(pOut->tbName, metaRsp.stbName);
code = queryCreateTableMetaFromMsg(pMetaMsg, true, &pOut->tbMeta);
pOut->ctbMeta.vgId = metaRsp.vgId;
pOut->ctbMeta.tableType = metaRsp.tableType;
pOut->ctbMeta.uid = metaRsp.tuid;
pOut->ctbMeta.suid = metaRsp.suid;
code = queryCreateTableMetaFromMsg(&metaRsp, true, &pOut->tbMeta);
} else {
SET_META_TYPE_TABLE(pOut->metaType);
strcpy(pOut->tbName, pMetaMsg->tbName);
code = queryCreateTableMetaFromMsg(pMetaMsg, (pMetaMsg->tableType == TSDB_SUPER_TABLE), &pOut->tbMeta);
strcpy(pOut->tbName, metaRsp.tbName);
code = queryCreateTableMetaFromMsg(&metaRsp, (metaRsp.tableType == TSDB_SUPER_TABLE), &pOut->tbMeta);
}
PROCESS_META_OVER:
if (code != 0) {
qError("failed to process table meta rsp since %s", terrstr());
}
tFreeSTableMetaRsp(&metaRsp);
return code;
}
void initQueryModuleMsgHandle() {
queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryBuildTableMetaReqMsg;
queryBuildMsg[TMSG_INDEX(TDMT_MND_STB_META)] = queryBuildTableMetaReqMsg;

39
source/libs/qworker/src/qworkerMsg.c
View File

@ -172,45 +172,54 @@ int32_t qwBuildAndSendDropRsp(void *connection, int32_t code) {
}
int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
int32_t numOfCols = 6;
int32_t msgSize = sizeof(SVShowTablesRsp) + sizeof(SSchema) * numOfCols;
int32_t numOfCols = 6;
SVShowTablesRsp showRsp = {0};
SVShowTablesRsp *pRsp = (SVShowTablesRsp *)rpcMallocCont(msgSize);
// showRsp.showId = 1;
showRsp.tableMeta.pSchemas = calloc(numOfCols, sizeof(SSchema));
if (showRsp.tableMeta.pSchemas == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
int32_t cols = 0;
SSchema *pSchema = pRsp->metaInfo.pSchema;
SSchema *pSchema = showRsp.tableMeta.pSchemas;
const SSchema *s = tGetTbnameColumnSchema();
*pSchema = createSchema(s->type, htonl(s->bytes), htonl(++cols), "name");
*pSchema = createSchema(s->type, s->bytes, ++cols, "name");
pSchema++;
int32_t type = TSDB_DATA_TYPE_TIMESTAMP;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "created");
*pSchema = createSchema(type, tDataTypes[type].bytes, ++cols, "created");
pSchema++;
type = TSDB_DATA_TYPE_SMALLINT;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "columns");
*pSchema = createSchema(type, tDataTypes[type].bytes, ++cols, "columns");
pSchema++;
*pSchema = createSchema(s->type, htonl(s->bytes), htonl(++cols), "stable");
*pSchema = createSchema(s->type, s->bytes, ++cols, "stable");
pSchema++;
type = TSDB_DATA_TYPE_BIGINT;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "uid");
*pSchema = createSchema(type, tDataTypes[type].bytes, ++cols, "uid");
pSchema++;
type = TSDB_DATA_TYPE_INT;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "vgId");
*pSchema = createSchema(type, tDataTypes[type].bytes, ++cols, "vgId");
assert(cols == numOfCols);
pRsp->metaInfo.numOfColumns = htonl(cols);
showRsp.tableMeta.numOfColumns = cols;
int32_t bufLen = tSerializeSShowRsp(NULL, 0, &showRsp);
void *pBuf = rpcMallocCont(bufLen);
tSerializeSShowRsp(pBuf, bufLen, &showRsp);
SRpcMsg rpcMsg = {
.handle = pMsg->handle,
.handle = pMsg->handle,
.ahandle = pMsg->ahandle,
.pCont = pRsp,
.contLen = msgSize,
.code = code,
.pCont = pBuf,
.contLen = bufLen,
.code = code,
};
rpcSendResponse(&rpcMsg);