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

enh: error handle in message encode and decode

This commit is contained in:
Hongze Cheng 2024-09-09 11:49:44 +08:00
parent 8012a1902b
commit a260a5f2f7
5 changed files with 716 additions and 542 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
source/dnode/mnode/impl/src/mndCompact.c
View File

@ -55,36 +55,44 @@ void tFreeCompactObj(SCompactObj *pCompact) {}
int32_t tSerializeSCompactObj(void *buf, int32_t bufLen, const SCompactObj *pObj) { int32_t tSerializeSCompactObj(void *buf, int32_t bufLen, const SCompactObj *pObj) {
SEncoder encoder = {0}; SEncoder encoder = {0};
int32_t code = 0;
int32_t lino;
int32_t tlen;
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1; TAOS_CHECK_EXIT(tStartEncode(&encoder));
TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->compactId));
if (tEncodeI32(&encoder, pObj->compactId) < 0) return -1; TAOS_CHECK_EXIT(tEncodeCStr(&encoder, pObj->dbname));
if (tEncodeCStr(&encoder, pObj->dbname) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI64(&encoder, pObj->startTime));
if (tEncodeI64(&encoder, pObj->startTime) < 0) return -1;
tEndEncode(&encoder); tEndEncode(&encoder);
int32_t tlen = encoder.pos; _exit:
if (code) {
tlen = code;
} else {
tlen = encoder.pos;
}
tEncoderClear(&encoder); tEncoderClear(&encoder);
return tlen; return tlen;
} }
int32_t tDeserializeSCompactObj(void *buf, int32_t bufLen, SCompactObj *pObj) { int32_t tDeserializeSCompactObj(void *buf, int32_t bufLen, SCompactObj *pObj) {
int8_t ex = 0; int32_t code = 0;
int32_t lino;
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
TAOS_CHECK_RETURN(tStartDecode(&decoder)); TAOS_CHECK_EXIT(tStartDecode(&decoder));
TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->compactId));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->compactId)); TAOS_CHECK_EXIT(tDecodeCStrTo(&decoder, pObj->dbname));
TAOS_CHECK_RETURN(tDecodeCStrTo(&decoder, pObj->dbname)); TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pObj->startTime));
TAOS_CHECK_RETURN(tDecodeI64(&decoder, &pObj->startTime));
tEndDecode(&decoder); tEndDecode(&decoder);
_exit:
tDecoderClear(&decoder); tDecoderClear(&decoder);
return 0; return code;
} }
SSdbRaw *mndCompactActionEncode(SCompactObj *pCompact) { SSdbRaw *mndCompactActionEncode(SCompactObj *pCompact) {

82
source/dnode/mnode/impl/src/mndCompactDetail.c
View File

@ -13,9 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "mndCompactDetail.h" #include "mndCompactDetail.h"
#include "mndTrans.h"
#include "mndShow.h"
#include "mndDb.h" #include "mndDb.h"
#include "mndShow.h"
#include "mndTrans.h"
#define MND_COMPACT_VER_NUMBER 1 #define MND_COMPACT_VER_NUMBER 1
@ -35,9 +35,7 @@ int32_t mndInitCompactDetail(SMnode *pMnode) {
return sdbSetTable(pMnode->pSdb, table); return sdbSetTable(pMnode->pSdb, table);
} }
void mndCleanupCompactDetail(SMnode *pMnode) { void mndCleanupCompactDetail(SMnode *pMnode) { mDebug("mnd compact detail cleanup"); }
mDebug("mnd compact detail cleanup");
}
int32_t mndRetrieveCompactDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) { int32_t mndRetrieveCompactDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
@ -49,7 +47,8 @@ int32_t mndRetrieveCompactDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
if (strlen(pShow->db) > 0) { if (strlen(pShow->db) > 0) {
sep = strchr(pShow->db, '.'); sep = strchr(pShow->db, '.');
if (sep && ((0 == strcmp(sep + 1, TSDB_INFORMATION_SCHEMA_DB) || (0 == strcmp(sep + 1, TSDB_PERFORMANCE_SCHEMA_DB))))) { if (sep &&
((0 == strcmp(sep + 1, TSDB_INFORMATION_SCHEMA_DB) || (0 == strcmp(sep + 1, TSDB_PERFORMANCE_SCHEMA_DB))))) {
sep++; sep++;
} else { } else {
pDb = mndAcquireDb(pMnode, pShow->db); pDb = mndAcquireDb(pMnode, pShow->db);
@ -94,53 +93,60 @@ int32_t mndRetrieveCompactDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
return numOfRows; return numOfRows;
} }
void tFreeCompactDetailObj(SCompactDetailObj *pCompact) { void tFreeCompactDetailObj(SCompactDetailObj *pCompact) {}
}
int32_t tSerializeSCompactDetailObj(void *buf, int32_t bufLen, const SCompactDetailObj *pObj) { int32_t tSerializeSCompactDetailObj(void *buf, int32_t bufLen, const SCompactDetailObj *pObj) {
SEncoder encoder = {0}; SEncoder encoder = {0};
int32_t code = 0;
int32_t lino;
int32_t tlen;
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1; TAOS_CHECK_EXIT(tStartEncode(&encoder));
TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->compactDetailId));
if (tEncodeI32(&encoder, pObj->compactDetailId) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->compactId));
if (tEncodeI32(&encoder, pObj->compactId) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->vgId));
if (tEncodeI32(&encoder, pObj->vgId) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->dnodeId));
if (tEncodeI32(&encoder, pObj->dnodeId) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->numberFileset));
if (tEncodeI32(&encoder, pObj->numberFileset) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->finished));
if (tEncodeI32(&encoder, pObj->finished) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI64(&encoder, pObj->startTime));
if (tEncodeI64(&encoder, pObj->startTime) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->newNumberFileset));
if (tEncodeI32(&encoder, pObj->newNumberFileset) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pObj->newFinished));
if (tEncodeI32(&encoder, pObj->newFinished) < 0) return -1;
tEndEncode(&encoder); tEndEncode(&encoder);
int32_t tlen = encoder.pos; _exit:
if (code) {
tlen = code;
} else {
tlen = encoder.pos;
}
tEncoderClear(&encoder); tEncoderClear(&encoder);
return tlen; return tlen;
} }
int32_t tDeserializeSCompactDetailObj(void *buf, int32_t bufLen, SCompactDetailObj *pObj) { int32_t tDeserializeSCompactDetailObj(void *buf, int32_t bufLen, SCompactDetailObj *pObj) {
int8_t ex = 0; int32_t code = 0;
int32_t lino;
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
TAOS_CHECK_RETURN(tStartDecode(&decoder)); TAOS_CHECK_EXIT(tStartDecode(&decoder));
TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->compactDetailId));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->compactDetailId)); TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->compactId));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->compactId)); TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->vgId));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->vgId)); TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->dnodeId));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->dnodeId)); TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->numberFileset));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->numberFileset)); TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->finished));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->finished)); TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pObj->startTime));
TAOS_CHECK_RETURN(tDecodeI64(&decoder, &pObj->startTime)); TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->newNumberFileset));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->newNumberFileset)); TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pObj->newFinished));
TAOS_CHECK_RETURN(tDecodeI32(&decoder, &pObj->newFinished));
tEndDecode(&decoder); tEndDecode(&decoder);
_exit:
tDecoderClear(&decoder); tDecoderClear(&decoder);
return 0; return code;
} }
SSdbRaw *mndCompactDetailActionEncode(SCompactDetailObj *pCompact) { SSdbRaw *mndCompactDetailActionEncode(SCompactDetailObj *pCompact) {
@ -181,7 +187,6 @@ SSdbRaw *mndCompactDetailActionEncode(SCompactDetailObj *pCompact) {
SDB_SET_BINARY(pRaw, dataPos, buf, tlen, OVER); SDB_SET_BINARY(pRaw, dataPos, buf, tlen, OVER);
SDB_SET_DATALEN(pRaw, dataPos, OVER); SDB_SET_DATALEN(pRaw, dataPos, OVER);
OVER: OVER:
taosMemoryFreeClear(buf); taosMemoryFreeClear(buf);
if (terrno != TSDB_CODE_SUCCESS) { if (terrno != TSDB_CODE_SUCCESS) {
@ -263,9 +268,8 @@ int32_t mndCompactDetailActionDelete(SSdb *pSdb, SCompactDetailObj *pCompact) {
} }
int32_t mndCompactDetailActionUpdate(SSdb *pSdb, SCompactDetailObj *pOldCompact, SCompactDetailObj *pNewCompact) { int32_t mndCompactDetailActionUpdate(SSdb *pSdb, SCompactDetailObj *pOldCompact, SCompactDetailObj *pNewCompact) {
mTrace("compact detail:%" PRId32 ", perform update action, old row:%p new row:%p", mTrace("compact detail:%" PRId32 ", perform update action, old row:%p new row:%p", pOldCompact->compactId,
pOldCompact->compactId, pOldCompact, pNewCompact); pOldCompact, pNewCompact);
pOldCompact->numberFileset = pNewCompact->numberFileset; pOldCompact->numberFileset = pNewCompact->numberFileset;
pOldCompact->finished = pNewCompact->finished; pOldCompact->finished = pNewCompact->finished;
@ -287,8 +291,8 @@ int32_t mndAddCompactDetailToTran(SMnode *pMnode, STrans *pTrans, SCompactObj* p
compactDetail.newNumberFileset = -1; compactDetail.newNumberFileset = -1;
compactDetail.newFinished = -1; compactDetail.newFinished = -1;
mInfo("compact:%d, add compact detail to trans, index:%d, vgId:%d, dnodeId:%d", mInfo("compact:%d, add compact detail to trans, index:%d, vgId:%d, dnodeId:%d", compactDetail.compactId,
compactDetail.compactId, compactDetail.compactDetailId, compactDetail.vgId, compactDetail.dnodeId); compactDetail.compactDetailId, compactDetail.vgId, compactDetail.dnodeId);
SSdbRaw *pVgRaw = mndCompactDetailActionEncode(&compactDetail); SSdbRaw *pVgRaw = mndCompactDetailActionEncode(&compactDetail);
if (pVgRaw == NULL) return -1; if (pVgRaw == NULL) return -1;

69
source/dnode/vnode/src/tq/tqMeta.c
View File

@ -16,64 +16,79 @@
#include "tq.h" #include "tq.h"
int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle) { int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle) {
if (tStartEncode(pEncoder) < 0) return -1; int32_t code = 0;
if (tEncodeCStr(pEncoder, pHandle->subKey) < 0) return -1; int32_t lino;
if (tEncodeI8(pEncoder, pHandle->fetchMeta) < 0) return -1;
if (tEncodeI64(pEncoder, pHandle->consumerId) < 0) return -1; TAOS_CHECK_EXIT(tStartEncode(pEncoder));
if (tEncodeI64(pEncoder, pHandle->snapshotVer) < 0) return -1; TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, pHandle->subKey));
if (tEncodeI32(pEncoder, pHandle->epoch) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pHandle->fetchMeta));
if (tEncodeI8(pEncoder, pHandle->execHandle.subType) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pHandle->consumerId));
TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pHandle->snapshotVer));
TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pHandle->epoch));
TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pHandle->execHandle.subType));
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
if (tEncodeCStr(pEncoder, pHandle->execHandle.execCol.qmsg) < 0) return -1; TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, pHandle->execHandle.execCol.qmsg));
} else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) { } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
int32_t size = taosHashGetSize(pHandle->execHandle.execDb.pFilterOutTbUid); int32_t size = taosHashGetSize(pHandle->execHandle.execDb.pFilterOutTbUid);
if (tEncodeI32(pEncoder, size) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(pEncoder, size));
void* pIter = NULL; void* pIter = NULL;
pIter = taosHashIterate(pHandle->execHandle.execDb.pFilterOutTbUid, pIter); pIter = taosHashIterate(pHandle->execHandle.execDb.pFilterOutTbUid, pIter);
while (pIter) { while (pIter) {
int64_t* tbUid = (int64_t*)taosHashGetKey(pIter, NULL); int64_t* tbUid = (int64_t*)taosHashGetKey(pIter, NULL);
if (tEncodeI64(pEncoder, *tbUid) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI64(pEncoder, *tbUid));
pIter = taosHashIterate(pHandle->execHandle.execDb.pFilterOutTbUid, pIter); pIter = taosHashIterate(pHandle->execHandle.execDb.pFilterOutTbUid, pIter);
} }
} else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) { } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
if (tEncodeI64(pEncoder, pHandle->execHandle.execTb.suid) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pHandle->execHandle.execTb.suid));
if (pHandle->execHandle.execTb.qmsg != NULL) { if (pHandle->execHandle.execTb.qmsg != NULL) {
if (tEncodeCStr(pEncoder, pHandle->execHandle.execTb.qmsg) < 0) return -1; TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, pHandle->execHandle.execTb.qmsg));
} }
} }
tEndEncode(pEncoder); tEndEncode(pEncoder);
_exit:
if (code) {
return code;
} else {
return pEncoder->pos; return pEncoder->pos;
} }
}
int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) { int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) {
if (tStartDecode(pDecoder) < 0) return -1; int32_t code = 0;
if (tDecodeCStrTo(pDecoder, pHandle->subKey) < 0) return -1; int32_t lino;
if (tDecodeI8(pDecoder, &pHandle->fetchMeta) < 0) return -1;
if (tDecodeI64(pDecoder, &pHandle->consumerId) < 0) return -1; TAOS_CHECK_EXIT(tStartDecode(pDecoder));
if (tDecodeI64(pDecoder, &pHandle->snapshotVer) < 0) return -1; TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, pHandle->subKey));
if (tDecodeI32(pDecoder, &pHandle->epoch) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pHandle->fetchMeta));
if (tDecodeI8(pDecoder, &pHandle->execHandle.subType) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pHandle->consumerId));
TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pHandle->snapshotVer));
TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pHandle->epoch));
TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pHandle->execHandle.subType));
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
if (tDecodeCStrAlloc(pDecoder, &pHandle->execHandle.execCol.qmsg) < 0) return -1; TAOS_CHECK_EXIT(tDecodeCStrAlloc(pDecoder, &pHandle->execHandle.execCol.qmsg));
} else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) { } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
pHandle->execHandle.execDb.pFilterOutTbUid = pHandle->execHandle.execDb.pFilterOutTbUid =
taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK); taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
if (pHandle->execHandle.execDb.pFilterOutTbUid == NULL) return -1; if (pHandle->execHandle.execDb.pFilterOutTbUid == NULL) {
TAOS_CHECK_EXIT(terrno);
}
int32_t size = 0; int32_t size = 0;
if (tDecodeI32(pDecoder, &size) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &size));
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
int64_t tbUid = 0; int64_t tbUid = 0;
if (tDecodeI64(pDecoder, &tbUid) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &tbUid));
if (taosHashPut(pHandle->execHandle.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0) != 0) return -1; TAOS_CHECK_EXIT(taosHashPut(pHandle->execHandle.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0));
} }
} else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) { } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
if (tDecodeI64(pDecoder, &pHandle->execHandle.execTb.suid) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pHandle->execHandle.execTb.suid));
if (!tDecodeIsEnd(pDecoder)) { if (!tDecodeIsEnd(pDecoder)) {
if (tDecodeCStrAlloc(pDecoder, &pHandle->execHandle.execTb.qmsg) < 0) return -1; TAOS_CHECK_EXIT(tDecodeCStrAlloc(pDecoder, &pHandle->execHandle.execTb.qmsg));
} }
} }
tEndDecode(pDecoder); tEndDecode(pDecoder);
return 0;
_exit:
return code;
} }
int32_t tqMetaDecodeCheckInfo(STqCheckInfo* info, void* pVal, int32_t vLen) { int32_t tqMetaDecodeCheckInfo(STqCheckInfo* info, void* pVal, int32_t vLen) {

75
source/libs/function/src/builtinsimpl.c
View File

@ -6437,63 +6437,74 @@ int32_t blockDistFunction(SqlFunctionCtx* pCtx) {
int32_t tSerializeBlockDistInfo(void* buf, int32_t bufLen, const STableBlockDistInfo* pInfo) { int32_t tSerializeBlockDistInfo(void* buf, int32_t bufLen, const STableBlockDistInfo* pInfo) {
SEncoder encoder = {0}; SEncoder encoder = {0};
int32_t code = 0;
int32_t lino;
int32_t tlen;
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1; TAOS_CHECK_EXIT(tStartEncode(&encoder));
if (tEncodeU32(&encoder, pInfo->rowSize) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU32(&encoder, pInfo->rowSize));
if (tEncodeU16(&encoder, pInfo->numOfFiles) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU16(&encoder, pInfo->numOfFiles));
if (tEncodeU32(&encoder, pInfo->numOfBlocks) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU32(&encoder, pInfo->numOfBlocks));
if (tEncodeU32(&encoder, pInfo->numOfTables) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU32(&encoder, pInfo->numOfTables));
if (tEncodeU64(&encoder, pInfo->totalSize) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU64(&encoder, pInfo->totalSize));
if (tEncodeU64(&encoder, pInfo->totalRows) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU64(&encoder, pInfo->totalRows));
if (tEncodeI32(&encoder, pInfo->maxRows) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pInfo->maxRows));
if (tEncodeI32(&encoder, pInfo->minRows) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pInfo->minRows));
if (tEncodeI32(&encoder, pInfo->defMaxRows) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pInfo->defMaxRows));
if (tEncodeI32(&encoder, pInfo->defMinRows) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pInfo->defMinRows));
if (tEncodeU32(&encoder, pInfo->numOfInmemRows) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU32(&encoder, pInfo->numOfInmemRows));
if (tEncodeU32(&encoder, pInfo->numOfSttRows) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU32(&encoder, pInfo->numOfSttRows));
if (tEncodeU32(&encoder, pInfo->numOfVgroups) < 0) return -1; TAOS_CHECK_EXIT(tEncodeU32(&encoder, pInfo->numOfVgroups));
for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) { for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) {
if (tEncodeI32(&encoder, pInfo->blockRowsHisto[i]) < 0) return -1; TAOS_CHECK_EXIT(tEncodeI32(&encoder, pInfo->blockRowsHisto[i]));
} }
tEndEncode(&encoder); tEndEncode(&encoder);
int32_t tlen = encoder.pos; _exit:
if (code) {
tlen = code;
} else {
tlen = encoder.pos;
}
tEncoderClear(&encoder); tEncoderClear(&encoder);
return tlen; return tlen;
} }
int32_t tDeserializeBlockDistInfo(void* buf, int32_t bufLen, STableBlockDistInfo* pInfo) { int32_t tDeserializeBlockDistInfo(void* buf, int32_t bufLen, STableBlockDistInfo* pInfo) {
SDecoder decoder = {0}; SDecoder decoder = {0};
int32_t code = 0;
int32_t lino;
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1; TAOS_CHECK_EXIT(tStartDecode(&decoder));
if (tDecodeU32(&decoder, &pInfo->rowSize) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU32(&decoder, &pInfo->rowSize));
if (tDecodeU16(&decoder, &pInfo->numOfFiles) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU16(&decoder, &pInfo->numOfFiles));
if (tDecodeU32(&decoder, &pInfo->numOfBlocks) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU32(&decoder, &pInfo->numOfBlocks));
if (tDecodeU32(&decoder, &pInfo->numOfTables) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU32(&decoder, &pInfo->numOfTables));
if (tDecodeU64(&decoder, &pInfo->totalSize) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU64(&decoder, &pInfo->totalSize));
if (tDecodeU64(&decoder, &pInfo->totalRows) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU64(&decoder, &pInfo->totalRows));
if (tDecodeI32(&decoder, &pInfo->maxRows) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pInfo->maxRows));
if (tDecodeI32(&decoder, &pInfo->minRows) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pInfo->minRows));
if (tDecodeI32(&decoder, &pInfo->defMaxRows) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pInfo->defMaxRows));
if (tDecodeI32(&decoder, &pInfo->defMinRows) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pInfo->defMinRows));
if (tDecodeU32(&decoder, &pInfo->numOfInmemRows) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU32(&decoder, &pInfo->numOfInmemRows));
if (tDecodeU32(&decoder, &pInfo->numOfSttRows) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU32(&decoder, &pInfo->numOfSttRows));
if (tDecodeU32(&decoder, &pInfo->numOfVgroups) < 0) return -1; TAOS_CHECK_EXIT(tDecodeU32(&decoder, &pInfo->numOfVgroups));
for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) { for (int32_t i = 0; i < tListLen(pInfo->blockRowsHisto); ++i) {
if (tDecodeI32(&decoder, &pInfo->blockRowsHisto[i]) < 0) return -1; TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pInfo->blockRowsHisto[i]));
} }
_exit:
tDecoderClear(&decoder); tDecoderClear(&decoder);
return 0; return code;
} }
int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {

946
source/libs/stream/src/streamMsg.c
View File

File diff suppressed because it is too large Load Diff