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

fix:offset error in tmq for primary key

This commit is contained in:
wangmm0220 2024-03-25 09:04:08 +08:00
parent 912cb09839
commit bd4be25316
2 changed files with 8 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/client/inc/clientInt.h
View File

@ -225,7 +225,6 @@ typedef struct {
} SMqRspObj;
typedef struct {
int8_t version;
int8_t resType;
char topic[TSDB_TOPIC_FNAME_LEN];
char db[TSDB_DB_FNAME_LEN];
@ -234,7 +233,6 @@ typedef struct {
} SMqMetaRspObj;
typedef struct {
int8_t version;
int8_t resType;
char topic[TSDB_TOPIC_FNAME_LEN];
char db[TSDB_DB_FNAME_LEN];

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

@ -8642,7 +8642,8 @@ void tFreeSMCreateStbRsp(SMCreateStbRsp *pRsp) {
}
int32_t tEncodeSTqOffsetVal(SEncoder *pEncoder, const STqOffsetVal *pOffsetVal) {
if (tEncodeI8(pEncoder, (TQ_OFFSET_VERSION << 4) | pOffsetVal->type) < 0) return -1;
int8_t type = pOffsetVal->type < 0 ? pOffsetVal->type : (TQ_OFFSET_VERSION << 4) | pOffsetVal->type;
if (tEncodeI8(pEncoder, type) < 0) return -1;
if (pOffsetVal->type == TMQ_OFFSET__SNAPSHOT_DATA || pOffsetVal->type == TMQ_OFFSET__SNAPSHOT_META) {
if (tEncodeI64(pEncoder, pOffsetVal->uid) < 0) return -1;
if (tEncodeI64(pEncoder, pOffsetVal->ts) < 0) return -1;
@ -8663,11 +8664,15 @@ int32_t tEncodeSTqOffsetVal(SEncoder *pEncoder, const STqOffsetVal *pOffsetVal)
int32_t tDecodeSTqOffsetVal(SDecoder *pDecoder, STqOffsetVal *pOffsetVal) {
if (tDecodeI8(pDecoder, &pOffsetVal->type) < 0) return -1;
int8_t offsetVersion = 0;
if (pOffsetVal->type > 0){
offsetVersion = (pOffsetVal->type >> 4);
pOffsetVal->type = pOffsetVal->type & 0x0F;
}
if (pOffsetVal->type == TMQ_OFFSET__SNAPSHOT_DATA || pOffsetVal->type == TMQ_OFFSET__SNAPSHOT_META) {
if (tDecodeI64(pDecoder, &pOffsetVal->uid) < 0) return -1;
if (tDecodeI64(pDecoder, &pOffsetVal->ts) < 0) return -1;
if ((pOffsetVal->type >> 4) >= TQ_OFFSET_VERSION) {
pOffsetVal->type = pOffsetVal->type & 0x0F;
if (offsetVersion >= TQ_OFFSET_VERSION) {
if (tDecodeI8(pDecoder, &pOffsetVal->primaryKey.type) < 0) return -1;
if (IS_VAR_DATA_TYPE(pOffsetVal->primaryKey.type)){
if (tDecodeBinaryAlloc32(pDecoder, &pOffsetVal->primaryKey.pData, &pOffsetVal->primaryKey.nData) < 0) return -1;