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

enh: add binary serialization method to node structure

This commit is contained in:
Xiaoyu Wang 2022-09-14 10:34:13 +08:00
parent 26a1bb437c
commit fbcef61d06
2 changed files with 46 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4718,9 +4718,8 @@ int32_t tSerializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq) {
if (tEncodeU64(&encoder, pReq->queryId) < 0) return -1; if (tEncodeU64(&encoder, pReq->queryId) < 0) return -1;
if (tEncodeU64(&encoder, pReq->taskId) < 0) return -1; if (tEncodeU64(&encoder, pReq->taskId) < 0) return -1;
if (tEncodeU32(&encoder, pReq->sqlLen) < 0) return -1; if (tEncodeU32(&encoder, pReq->sqlLen) < 0) return -1;
if (tEncodeU32(&encoder, pReq->phyLen) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->sql) < 0) return -1; if (tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->msg) < 0) return -1; if (tEncodeBinary(&encoder, pReq->msg, pReq->phyLen) < 0) return -1;
tEndEncode(&encoder); tEndEncode(&encoder);
int32_t tlen = encoder.pos; int32_t tlen = encoder.pos;
@ -4750,13 +4749,12 @@ int32_t tDeserializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq) {
if (tDecodeU64(&decoder, &pReq->queryId) < 0) return -1; if (tDecodeU64(&decoder, &pReq->queryId) < 0) return -1;
if (tDecodeU64(&decoder, &pReq->taskId) < 0) return -1; if (tDecodeU64(&decoder, &pReq->taskId) < 0) return -1;
if (tDecodeU32(&decoder, &pReq->sqlLen) < 0) return -1; if (tDecodeU32(&decoder, &pReq->sqlLen) < 0) return -1;
if (tDecodeU32(&decoder, &pReq->phyLen) < 0) return -1;
pReq->sql = taosMemoryCalloc(1, pReq->sqlLen + 1); pReq->sql = taosMemoryCalloc(1, pReq->sqlLen + 1);
if (NULL == pReq->sql) return -1; if (NULL == pReq->sql) return -1;
pReq->msg = taosMemoryCalloc(1, pReq->phyLen + 1);
if (NULL == pReq->msg) return -1;
if (tDecodeCStrTo(&decoder, pReq->sql) < 0) return -1; if (tDecodeCStrTo(&decoder, pReq->sql) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->msg) < 0) return -1; uint64_t msgLen = 0;
if (tDecodeBinaryAlloc(&decoder, (void **)&pReq->msg, &msgLen) < 0) return -1;
pReq->phyLen = msgLen;
tEndDecode(&decoder); tEndDecode(&decoder);

47
source/libs/nodes/src/nodesMsgFuncs.c
View File

@ -254,6 +254,11 @@ static int32_t tlvDecodeDynBinary(STlv* pTlv, void** pValue) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tlvDecodeBinary(STlv* pTlv, void* pValue) {
memcpy(pValue, pTlv->value, pTlv->len);
return TSDB_CODE_SUCCESS;
}
static int32_t tlvDecodeObjFromTlv(STlv* pTlv, FToObject func, void* pObj) { static int32_t tlvDecodeObjFromTlv(STlv* pTlv, FToObject func, void* pObj) {
STlvDecoder decoder = {.bufSize = pTlv->len, .offset = 0, .pBuf = pTlv->value}; STlvDecoder decoder = {.bufSize = pTlv->len, .offset = 0, .pBuf = pTlv->value};
return func(&decoder, pObj); return func(&decoder, pObj);
@ -469,7 +474,15 @@ static int32_t msgToColumnNode(STlvDecoder* pDecoder, void* pObj) {
return code; return code;
} }
enum { VALUE_CODE_EXPR_BASE = 1, VALUE_CODE_LITERAL, VALUE_CODE_IS_NULL, VALUE_CODE_DATUM }; enum {
VALUE_CODE_EXPR_BASE = 1,
VALUE_CODE_LITERAL,
VALUE_CODE_IS_DURATION,
VALUE_CODE_TRANSLATE,
VALUE_CODE_NOT_RESERVED,
VALUE_CODE_IS_NULL,
VALUE_CODE_DATUM
};
static int32_t datumToMsg(const void* pObj, STlvEncoder* pEncoder) { static int32_t datumToMsg(const void* pObj, STlvEncoder* pEncoder) {
const SValueNode* pNode = (const SValueNode*)pObj; const SValueNode* pNode = (const SValueNode*)pObj;
@ -524,9 +537,18 @@ static int32_t valueNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
code = tlvEncodeCStr(pEncoder, VALUE_CODE_LITERAL, pNode->literal); code = tlvEncodeCStr(pEncoder, VALUE_CODE_LITERAL, pNode->literal);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, VALUE_CODE_IS_NULL, pNode->isNull); code = tlvEncodeBool(pEncoder, VALUE_CODE_IS_DURATION, pNode->isDuration);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, VALUE_CODE_TRANSLATE, pNode->translate);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, VALUE_CODE_NOT_RESERVED, pNode->notReserved);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, VALUE_CODE_IS_NULL, pNode->isNull);
}
if (TSDB_CODE_SUCCESS == code && !pNode->isNull) {
code = datumToMsg(pNode, pEncoder); code = datumToMsg(pNode, pEncoder);
} }
@ -590,12 +612,18 @@ static int32_t msgToDatum(STlv* pTlv, void* pObj) {
break; break;
case TSDB_DATA_TYPE_NCHAR: case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARCHAR: case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY: case TSDB_DATA_TYPE_VARBINARY: {
code = tlvDecodeDynBinary(pTlv, (void**)&pNode->datum.p); pNode->datum.p = taosMemoryCalloc(1, pNode->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
if (NULL == pNode->datum.p) {
code = TSDB_CODE_OUT_OF_MEMORY;
break;
}
code = tlvDecodeBinary(pTlv, pNode->datum.p);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
varDataSetLen(pNode->datum.p, pNode->node.resType.bytes - VARSTR_HEADER_SIZE); varDataSetLen(pNode->datum.p, pTlv->len - VARSTR_HEADER_SIZE);
} }
break; break;
}
case TSDB_DATA_TYPE_JSON: case TSDB_DATA_TYPE_JSON:
code = tlvDecodeDynBinary(pTlv, (void**)&pNode->datum.p); code = tlvDecodeDynBinary(pTlv, (void**)&pNode->datum.p);
break; break;
@ -622,6 +650,15 @@ static int32_t msgToValueNode(STlvDecoder* pDecoder, void* pObj) {
case VALUE_CODE_LITERAL: case VALUE_CODE_LITERAL:
code = tlvDecodeCStrP(pTlv, &pNode->literal); code = tlvDecodeCStrP(pTlv, &pNode->literal);
break; break;
case VALUE_CODE_IS_DURATION:
code = tlvDecodeBool(pTlv, &pNode->isDuration);
break;
case VALUE_CODE_TRANSLATE:
code = tlvDecodeBool(pTlv, &pNode->translate);
break;
case VALUE_CODE_NOT_RESERVED:
code = tlvDecodeBool(pTlv, &pNode->notReserved);
break;
case VALUE_CODE_IS_NULL: case VALUE_CODE_IS_NULL:
code = tlvDecodeBool(pTlv, &pNode->isNull); code = tlvDecodeBool(pTlv, &pNode->isNull);
break; break;