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

Add support for updating column compression in alter table

This commit is contained in:
Yihao Deng 2024-04-23 10:10:31 +00:00
parent b244344fb3
commit 1cf109604c
2 changed files with 105 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -199,6 +199,32 @@ static char* buildAlterSTableJson(void* alterData, int32_t alterDataLen) {
cJSON_AddItemToObject(json, "colNewName", colNewName); cJSON_AddItemToObject(json, "colNewName", colNewName);
break; break;
} }
case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {
TAOS_FIELD* field = taosArrayGet(req.pFields, 0);
cJSON* colName = cJSON_CreateString(field->name);
cJSON_AddItemToObject(json, "colName", colName);
uint8_t encode = COMPRESS_L1_TYPE_U32(field->bytes);
if (encode != 0) {
const char* encodeStr = columnEncodeStr(encode);
cJSON* encodeJson = cJSON_CreateString(encodeStr);
cJSON_AddItemToObject(json, "encode", encodeJson);
break;
}
uint8_t compress = COMPRESS_L2_TYPE_U32(field->bytes);
if (compress != 0) {
const char* compressStr = columnCompressStr(compress);
cJSON* compressJson = cJSON_CreateString(compressStr);
cJSON_AddItemToObject(json, "compress", compressJson);
break;
}
uint8_t level = COMPRESS_L2_TYPE_LEVEL_U32(field->bytes);
if (level != 0) {
const char* levelStr = columnLevelStr(level);
cJSON* levelJson = cJSON_CreateString(levelStr);
cJSON_AddItemToObject(json, "level", levelJson);
break;
}
}
default: default:
break; break;
} }
@ -568,6 +594,32 @@ static char* processAlterTable(SMqMetaRsp* metaRsp) {
cJSON_AddItemToObject(json, "colValueNull", isNullCJson); cJSON_AddItemToObject(json, "colValueNull", isNullCJson);
break; break;
} }
case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {
cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);
cJSON_AddItemToObject(json, "colName", colName);
uint8_t encode = COMPRESS_L1_TYPE_U32(vAlterTbReq.compress);
if (encode != 0) {
const char* encodeStr = columnEncodeStr(encode);
cJSON* encodeJson = cJSON_CreateString(encodeStr);
cJSON_AddItemToObject(json, "encode", encodeJson);
break;
}
uint8_t compress = COMPRESS_L2_TYPE_U32(vAlterTbReq.compress);
if (compress != 0) {
const char* compressStr = columnCompressStr(compress);
cJSON* compressJson = cJSON_CreateString(compressStr);
cJSON_AddItemToObject(json, "compress", compressJson);
break;
}
uint8_t level = COMPRESS_L2_TYPE_LEVEL_U32(vAlterTbReq.compress);
if (level != 0) {
const char* levelStr = columnLevelStr(level);
cJSON* levelJson = cJSON_CreateString(levelStr);
cJSON_AddItemToObject(json, "level", levelJson);
break;
}
break;
}
default: default:
break; break;
} }

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

@ -69,7 +69,7 @@
static int32_t tDecodeSVAlterTbReqCommon(SDecoder *pDecoder, SVAlterTbReq *pReq); static int32_t tDecodeSVAlterTbReqCommon(SDecoder *pDecoder, SVAlterTbReq *pReq);
static int32_t tDecodeSBatchDeleteReqCommon(SDecoder *pDecoder, SBatchDeleteReq *pReq); static int32_t tDecodeSBatchDeleteReqCommon(SDecoder *pDecoder, SBatchDeleteReq *pReq);
static int32_t tEncodeTableTSMAInfoRsp(SEncoder *pEncoder, const STableTSMAInfoRsp *pRsp); static int32_t tEncodeTableTSMAInfoRsp(SEncoder *pEncoder, const STableTSMAInfoRsp *pRsp);
static int32_t tDecodeTableTSMAInfoRsp(SDecoder* pDecoder, STableTSMAInfoRsp* pRsp); static int32_t tDecodeTableTSMAInfoRsp(SDecoder *pDecoder, STableTSMAInfoRsp *pRsp);
int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) { int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
if (pMsg == NULL) { if (pMsg == NULL) {
@ -895,8 +895,8 @@ int32_t tSerializeSMCreateSmaReq(void *buf, int32_t bufLen, SMCreateSmaReq *pReq
if (tEncodeI64(&encoder, pReq->normSourceTbUid) < 0) return -1; if (tEncodeI64(&encoder, pReq->normSourceTbUid) < 0) return -1;
if (tEncodeI32(&encoder, taosArrayGetSize(pReq->pVgroupVerList)) < 0) return -1; if (tEncodeI32(&encoder, taosArrayGetSize(pReq->pVgroupVerList)) < 0) return -1;
for(int32_t i = 0; i < taosArrayGetSize(pReq->pVgroupVerList); ++i) { for (int32_t i = 0; i < taosArrayGetSize(pReq->pVgroupVerList); ++i) {
SVgroupVer* p = taosArrayGet(pReq->pVgroupVerList, i); SVgroupVer *p = taosArrayGet(pReq->pVgroupVerList, i);
if (tEncodeI32(&encoder, p->vgId) < 0) return -1; if (tEncodeI32(&encoder, p->vgId) < 0) return -1;
if (tEncodeI64(&encoder, p->ver) < 0) return -1; if (tEncodeI64(&encoder, p->ver) < 0) return -1;
} }
@ -8000,7 +8000,7 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
} }
} }
if (!tDecodeIsEnd(&decoder)) { if (!tDecodeIsEnd(&decoder)) {
if (tDecodeI64(&decoder, &pReq->smaId)< 0) return -1; if (tDecodeI64(&decoder, &pReq->smaId) < 0) return -1;
} }
tEndDecode(&decoder); tEndDecode(&decoder);
@ -8445,8 +8445,8 @@ static int32_t tDecodeSVDropTbRsp(SDecoder *pCoder, SVDropTbRsp *pReq) {
} }
int32_t tEncodeSVDropTbBatchReq(SEncoder *pCoder, const SVDropTbBatchReq *pReq) { int32_t tEncodeSVDropTbBatchReq(SEncoder *pCoder, const SVDropTbBatchReq *pReq) {
int32_t nReqs = taosArrayGetSize(pReq->pArray); int32_t nReqs = taosArrayGetSize(pReq->pArray);
SVDropTbReq *pDropTbReq; SVDropTbReq *pDropTbReq;
if (tStartEncode(pCoder) < 0) return -1; if (tStartEncode(pCoder) < 0) return -1;
@ -8709,6 +8709,7 @@ int32_t tEncodeSVAlterTbReq(SEncoder *pEncoder, const SVAlterTbReq *pReq) {
} }
break; break;
case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS:
if (tEncodeCStr(pEncoder, pReq->colName) < 0) return -1;
if (tEncodeU32(pEncoder, pReq->compress) < 0) return -1; if (tEncodeU32(pEncoder, pReq->compress) < 0) return -1;
break; break;
default: default:
@ -8763,6 +8764,7 @@ static int32_t tDecodeSVAlterTbReqCommon(SDecoder *pDecoder, SVAlterTbReq *pReq)
} }
break; break;
case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS:
if (tDecodeCStr(pDecoder, &pReq->colName) < 0) return -1;
if (tDecodeU32(pDecoder, &pReq->compress) < 0) return -1; if (tDecodeU32(pDecoder, &pReq->compress) < 0) return -1;
break; break;
default: default:
@ -9200,7 +9202,7 @@ int32_t tEncodeMqDataRspCommon(SEncoder *pEncoder, const SMqDataRspCommon *pRsp)
int32_t tEncodeMqDataRsp(SEncoder *pEncoder, const void *pRsp) { int32_t tEncodeMqDataRsp(SEncoder *pEncoder, const void *pRsp) {
if (tEncodeMqDataRspCommon(pEncoder, pRsp) < 0) return -1; if (tEncodeMqDataRspCommon(pEncoder, pRsp) < 0) return -1;
if (tEncodeI64(pEncoder, ((SMqDataRsp*)pRsp)->sleepTime) < 0) return -1; if (tEncodeI64(pEncoder, ((SMqDataRsp *)pRsp)->sleepTime) < 0) return -1;
return 0; return 0;
} }
@ -9253,7 +9255,7 @@ int32_t tDecodeMqDataRspCommon(SDecoder *pDecoder, SMqDataRspCommon *pRsp) {
int32_t tDecodeMqDataRsp(SDecoder *pDecoder, void *pRsp) { int32_t tDecodeMqDataRsp(SDecoder *pDecoder, void *pRsp) {
if (tDecodeMqDataRspCommon(pDecoder, pRsp) < 0) return -1; if (tDecodeMqDataRspCommon(pDecoder, pRsp) < 0) return -1;
if (!tDecodeIsEnd(pDecoder)) { if (!tDecodeIsEnd(pDecoder)) {
if (tDecodeI64(pDecoder, &((SMqDataRsp*)pRsp)->sleepTime) < 0) return -1; if (tDecodeI64(pDecoder, &((SMqDataRsp *)pRsp)->sleepTime) < 0) return -1;
} }
return 0; return 0;
@ -9272,9 +9274,7 @@ static void tDeleteMqDataRspCommon(void *rsp) {
tOffsetDestroy(&pRsp->rspOffset); tOffsetDestroy(&pRsp->rspOffset);
} }
void tDeleteMqDataRsp(void *rsp) { void tDeleteMqDataRsp(void *rsp) { tDeleteMqDataRspCommon(rsp); }
tDeleteMqDataRspCommon(rsp);
}
int32_t tEncodeSTaosxRsp(SEncoder *pEncoder, const void *rsp) { int32_t tEncodeSTaosxRsp(SEncoder *pEncoder, const void *rsp) {
if (tEncodeMqDataRspCommon(pEncoder, rsp) < 0) return -1; if (tEncodeMqDataRspCommon(pEncoder, rsp) < 0) return -1;
@ -9300,7 +9300,7 @@ int32_t tDecodeSTaosxRsp(SDecoder *pDecoder, void *rsp) {
pRsp->createTableLen = taosArrayInit(pRsp->createTableNum, sizeof(int32_t)); pRsp->createTableLen = taosArrayInit(pRsp->createTableNum, sizeof(int32_t));
pRsp->createTableReq = taosArrayInit(pRsp->createTableNum, sizeof(void *)); pRsp->createTableReq = taosArrayInit(pRsp->createTableNum, sizeof(void *));
for (int32_t i = 0; i < pRsp->createTableNum; i++) { for (int32_t i = 0; i < pRsp->createTableNum; i++) {
void * pCreate = NULL; void *pCreate = NULL;
uint64_t len = 0; uint64_t len = 0;
if (tDecodeBinaryAlloc(pDecoder, &pCreate, &len) < 0) return -1; if (tDecodeBinaryAlloc(pDecoder, &pCreate, &len) < 0) return -1;
int32_t l = (int32_t)len; int32_t l = (int32_t)len;
@ -10112,7 +10112,7 @@ void setFieldWithOptions(SFieldWithOptions *fieldWithOptions, SField *field) {
fieldWithOptions->type = field->type; fieldWithOptions->type = field->type;
strncpy(fieldWithOptions->name, field->name, TSDB_COL_NAME_LEN); strncpy(fieldWithOptions->name, field->name, TSDB_COL_NAME_LEN);
} }
int32_t tSerializeTableTSMAInfoReq(void* buf, int32_t bufLen, const STableTSMAInfoReq* pReq) { int32_t tSerializeTableTSMAInfoReq(void *buf, int32_t bufLen, const STableTSMAInfoReq *pReq) {
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
@ -10127,13 +10127,13 @@ int32_t tSerializeTableTSMAInfoReq(void* buf, int32_t bufLen, const STableTSMAIn
return tlen; return tlen;
} }
int32_t tDeserializeTableTSMAInfoReq(void* buf, int32_t bufLen, STableTSMAInfoReq* pReq) { int32_t tDeserializeTableTSMAInfoReq(void *buf, int32_t bufLen, STableTSMAInfoReq *pReq) {
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1; if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1; if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
if (tDecodeI8(&decoder, (uint8_t*)&pReq->fetchingWithTsmaName) < 0) return -1; if (tDecodeI8(&decoder, (uint8_t *)&pReq->fetchingWithTsmaName) < 0) return -1;
tEndDecode(&decoder); tEndDecode(&decoder);
@ -10141,7 +10141,7 @@ int32_t tDeserializeTableTSMAInfoReq(void* buf, int32_t bufLen, STableTSMAInfoRe
return 0; return 0;
} }
static int32_t tEncodeTableTSMAInfo(SEncoder* pEncoder, const STableTSMAInfo* pTsmaInfo) { static int32_t tEncodeTableTSMAInfo(SEncoder *pEncoder, const STableTSMAInfo *pTsmaInfo) {
if (tEncodeCStr(pEncoder, pTsmaInfo->name) < 0) return -1; if (tEncodeCStr(pEncoder, pTsmaInfo->name) < 0) return -1;
if (tEncodeU64(pEncoder, pTsmaInfo->tsmaId) < 0) return -1; if (tEncodeU64(pEncoder, pTsmaInfo->tsmaId) < 0) return -1;
if (tEncodeCStr(pEncoder, pTsmaInfo->tb) < 0) return -1; if (tEncodeCStr(pEncoder, pTsmaInfo->tb) < 0) return -1;
@ -10158,7 +10158,7 @@ static int32_t tEncodeTableTSMAInfo(SEncoder* pEncoder, const STableTSMAInfo* pT
int32_t size = pTsmaInfo->pFuncs ? pTsmaInfo->pFuncs->size : 0; int32_t size = pTsmaInfo->pFuncs ? pTsmaInfo->pFuncs->size : 0;
if (tEncodeI32(pEncoder, size) < 0) return -1; if (tEncodeI32(pEncoder, size) < 0) return -1;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
STableTSMAFuncInfo* pFuncInfo = taosArrayGet(pTsmaInfo->pFuncs, i); STableTSMAFuncInfo *pFuncInfo = taosArrayGet(pTsmaInfo->pFuncs, i);
if (tEncodeI32(pEncoder, pFuncInfo->funcId) < 0) return -1; if (tEncodeI32(pEncoder, pFuncInfo->funcId) < 0) return -1;
if (tEncodeI16(pEncoder, pFuncInfo->colId) < 0) return -1; if (tEncodeI16(pEncoder, pFuncInfo->colId) < 0) return -1;
} }
@ -10166,13 +10166,13 @@ static int32_t tEncodeTableTSMAInfo(SEncoder* pEncoder, const STableTSMAInfo* pT
size = pTsmaInfo->pTags ? pTsmaInfo->pTags->size : 0; size = pTsmaInfo->pTags ? pTsmaInfo->pTags->size : 0;
if (tEncodeI32(pEncoder, size) < 0) return -1; if (tEncodeI32(pEncoder, size) < 0) return -1;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
const SSchema* pSchema = taosArrayGet(pTsmaInfo->pTags, i); const SSchema *pSchema = taosArrayGet(pTsmaInfo->pTags, i);
if (tEncodeSSchema(pEncoder, pSchema) < 0) return -1; if (tEncodeSSchema(pEncoder, pSchema) < 0) return -1;
} }
size = pTsmaInfo->pUsedCols ? pTsmaInfo->pUsedCols->size : 0; size = pTsmaInfo->pUsedCols ? pTsmaInfo->pUsedCols->size : 0;
if (tEncodeI32(pEncoder, size) < 0) return -1; if (tEncodeI32(pEncoder, size) < 0) return -1;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
const SSchema* pSchema = taosArrayGet(pTsmaInfo->pUsedCols, i); const SSchema *pSchema = taosArrayGet(pTsmaInfo->pUsedCols, i);
if (tEncodeSSchema(pEncoder, pSchema) < 0) return -1; if (tEncodeSSchema(pEncoder, pSchema) < 0) return -1;
} }
@ -10185,7 +10185,7 @@ static int32_t tEncodeTableTSMAInfo(SEncoder* pEncoder, const STableTSMAInfo* pT
return 0; return 0;
} }
static int32_t tDecodeTableTSMAInfo(SDecoder* pDecoder, STableTSMAInfo* pTsmaInfo) { static int32_t tDecodeTableTSMAInfo(SDecoder *pDecoder, STableTSMAInfo *pTsmaInfo) {
if (tDecodeCStrTo(pDecoder, pTsmaInfo->name) < 0) return -1; if (tDecodeCStrTo(pDecoder, pTsmaInfo->name) < 0) return -1;
if (tDecodeU64(pDecoder, &pTsmaInfo->tsmaId) < 0) return -1; if (tDecodeU64(pDecoder, &pTsmaInfo->tsmaId) < 0) return -1;
if (tDecodeCStrTo(pDecoder, pTsmaInfo->tb) < 0) return -1; if (tDecodeCStrTo(pDecoder, pTsmaInfo->tb) < 0) return -1;
@ -10217,7 +10217,7 @@ static int32_t tDecodeTableTSMAInfo(SDecoder* pDecoder, STableTSMAInfo* pTsmaInf
if (!pTsmaInfo->pTags) return -1; if (!pTsmaInfo->pTags) return -1;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
SSchema schema = {0}; SSchema schema = {0};
if(tDecodeSSchema(pDecoder, &schema) < 0) return -1; if (tDecodeSSchema(pDecoder, &schema) < 0) return -1;
taosArrayPush(pTsmaInfo->pTags, &schema); taosArrayPush(pTsmaInfo->pTags, &schema);
} }
} }
@ -10237,7 +10237,7 @@ static int32_t tDecodeTableTSMAInfo(SDecoder* pDecoder, STableTSMAInfo* pTsmaInf
if (tDecodeI64(pDecoder, &pTsmaInfo->reqTs) < 0) return -1; if (tDecodeI64(pDecoder, &pTsmaInfo->reqTs) < 0) return -1;
if (tDecodeI64(pDecoder, &pTsmaInfo->rspTs) < 0) return -1; if (tDecodeI64(pDecoder, &pTsmaInfo->rspTs) < 0) return -1;
if (tDecodeI64(pDecoder, &pTsmaInfo->delayDuration) < 0) return -1; if (tDecodeI64(pDecoder, &pTsmaInfo->delayDuration) < 0) return -1;
if (tDecodeI8(pDecoder, (int8_t*)&pTsmaInfo->fillHistoryFinished) < 0) return -1; if (tDecodeI8(pDecoder, (int8_t *)&pTsmaInfo->fillHistoryFinished) < 0) return -1;
return 0; return 0;
} }
@ -10245,13 +10245,13 @@ static int32_t tEncodeTableTSMAInfoRsp(SEncoder *pEncoder, const STableTSMAInfoR
int32_t size = pRsp->pTsmas ? pRsp->pTsmas->size : 0; int32_t size = pRsp->pTsmas ? pRsp->pTsmas->size : 0;
if (tEncodeI32(pEncoder, size) < 0) return -1; if (tEncodeI32(pEncoder, size) < 0) return -1;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
STableTSMAInfo* pInfo = taosArrayGetP(pRsp->pTsmas, i); STableTSMAInfo *pInfo = taosArrayGetP(pRsp->pTsmas, i);
if (tEncodeTableTSMAInfo(pEncoder, pInfo) < 0) return -1; if (tEncodeTableTSMAInfo(pEncoder, pInfo) < 0) return -1;
} }
return 0; return 0;
} }
static int32_t tDecodeTableTSMAInfoRsp(SDecoder* pDecoder, STableTSMAInfoRsp* pRsp) { static int32_t tDecodeTableTSMAInfoRsp(SDecoder *pDecoder, STableTSMAInfoRsp *pRsp) {
int32_t size = 0; int32_t size = 0;
if (tDecodeI32(pDecoder, &size) < 0) return -1; if (tDecodeI32(pDecoder, &size) < 0) return -1;
if (size <= 0) return 0; if (size <= 0) return 0;
@ -10266,7 +10266,7 @@ static int32_t tDecodeTableTSMAInfoRsp(SDecoder* pDecoder, STableTSMAInfoRsp* pR
return 0; return 0;
} }
int32_t tSerializeTableTSMAInfoRsp(void* buf, int32_t bufLen, const STableTSMAInfoRsp* pRsp) { int32_t tSerializeTableTSMAInfoRsp(void *buf, int32_t bufLen, const STableTSMAInfoRsp *pRsp) {
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
@ -10280,7 +10280,7 @@ int32_t tSerializeTableTSMAInfoRsp(void* buf, int32_t bufLen, const STableTSMAIn
return tlen; return tlen;
} }
int32_t tDeserializeTableTSMAInfoRsp(void* buf, int32_t bufLen, STableTSMAInfoRsp* pRsp) { int32_t tDeserializeTableTSMAInfoRsp(void *buf, int32_t bufLen, STableTSMAInfoRsp *pRsp) {
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
@ -10293,7 +10293,7 @@ int32_t tDeserializeTableTSMAInfoRsp(void* buf, int32_t bufLen, STableTSMAInfoRs
return 0; return 0;
} }
void tFreeTableTSMAInfo(void* p) { void tFreeTableTSMAInfo(void *p) {
STableTSMAInfo *pTsmaInfo = p; STableTSMAInfo *pTsmaInfo = p;
if (pTsmaInfo) { if (pTsmaInfo) {
taosArrayDestroy(pTsmaInfo->pFuncs); taosArrayDestroy(pTsmaInfo->pFuncs);
@ -10303,20 +10303,20 @@ void tFreeTableTSMAInfo(void* p) {
} }
} }
void tFreeAndClearTableTSMAInfo(void* p) { void tFreeAndClearTableTSMAInfo(void *p) {
STableTSMAInfo* pTsmaInfo = (STableTSMAInfo*)p; STableTSMAInfo *pTsmaInfo = (STableTSMAInfo *)p;
if (pTsmaInfo) { if (pTsmaInfo) {
tFreeTableTSMAInfo(pTsmaInfo); tFreeTableTSMAInfo(pTsmaInfo);
taosMemoryFree(pTsmaInfo); taosMemoryFree(pTsmaInfo);
} }
} }
int32_t tCloneTbTSMAInfo(STableTSMAInfo* pInfo, STableTSMAInfo** pRes) { int32_t tCloneTbTSMAInfo(STableTSMAInfo *pInfo, STableTSMAInfo **pRes) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
if (NULL == pInfo) { if (NULL == pInfo) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
STableTSMAInfo* pRet = taosMemoryCalloc(1, sizeof(STableTSMAInfo)); STableTSMAInfo *pRet = taosMemoryCalloc(1, sizeof(STableTSMAInfo));
if (!pRet) return TSDB_CODE_OUT_OF_MEMORY; if (!pRet) return TSDB_CODE_OUT_OF_MEMORY;
*pRet = *pInfo; *pRet = *pInfo;
@ -10355,7 +10355,7 @@ static int32_t tEncodeStreamProgressReq(SEncoder *pEncoder, const SStreamProgres
return 0; return 0;
} }
int32_t tSerializeStreamProgressReq(void* buf, int32_t bufLen, const SStreamProgressReq* pReq) { int32_t tSerializeStreamProgressReq(void *buf, int32_t bufLen, const SStreamProgressReq *pReq) {
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
@ -10369,7 +10369,7 @@ int32_t tSerializeStreamProgressReq(void* buf, int32_t bufLen, const SStreamProg
return tlen; return tlen;
} }
static int32_t tDecodeStreamProgressReq(SDecoder* pDecoder, SStreamProgressReq* pReq) { static int32_t tDecodeStreamProgressReq(SDecoder *pDecoder, SStreamProgressReq *pReq) {
if (tDecodeI64(pDecoder, &pReq->streamId) < 0) return -1; if (tDecodeI64(pDecoder, &pReq->streamId) < 0) return -1;
if (tDecodeI32(pDecoder, &pReq->vgId) < 0) return -1; if (tDecodeI32(pDecoder, &pReq->vgId) < 0) return -1;
if (tDecodeI32(pDecoder, &pReq->fetchIdx) < 0) return -1; if (tDecodeI32(pDecoder, &pReq->fetchIdx) < 0) return -1;
@ -10377,7 +10377,7 @@ static int32_t tDecodeStreamProgressReq(SDecoder* pDecoder, SStreamProgressReq*
return 0; return 0;
} }
int32_t tDeserializeStreamProgressReq(void* buf, int32_t bufLen, SStreamProgressReq* pReq) { int32_t tDeserializeStreamProgressReq(void *buf, int32_t bufLen, SStreamProgressReq *pReq) {
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, (char *)buf, bufLen); tDecoderInit(&decoder, (char *)buf, bufLen);
@ -10390,7 +10390,7 @@ int32_t tDeserializeStreamProgressReq(void* buf, int32_t bufLen, SStreamProgress
return 0; return 0;
} }
static int32_t tEncodeStreamProgressRsp(SEncoder* pEncoder, const SStreamProgressRsp* pRsp) { static int32_t tEncodeStreamProgressRsp(SEncoder *pEncoder, const SStreamProgressRsp *pRsp) {
if (tEncodeI64(pEncoder, pRsp->streamId) < 0) return -1; if (tEncodeI64(pEncoder, pRsp->streamId) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->vgId) < 0) return -1; if (tEncodeI32(pEncoder, pRsp->vgId) < 0) return -1;
if (tEncodeI8(pEncoder, pRsp->fillHisFinished) < 0) return -1; if (tEncodeI8(pEncoder, pRsp->fillHisFinished) < 0) return -1;
@ -10400,7 +10400,7 @@ static int32_t tEncodeStreamProgressRsp(SEncoder* pEncoder, const SStreamProgres
return 0; return 0;
} }
int32_t tSerializeStreamProgressRsp(void* buf, int32_t bufLen, const SStreamProgressRsp* pRsp) { int32_t tSerializeStreamProgressRsp(void *buf, int32_t bufLen, const SStreamProgressRsp *pRsp) {
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
@ -10414,17 +10414,17 @@ int32_t tSerializeStreamProgressRsp(void* buf, int32_t bufLen, const SStreamProg
return tlen; return tlen;
} }
static int32_t tDecodeStreamProgressRsp(SDecoder* pDecoder, SStreamProgressRsp* pRsp) { static int32_t tDecodeStreamProgressRsp(SDecoder *pDecoder, SStreamProgressRsp *pRsp) {
if (tDecodeI64(pDecoder, &pRsp->streamId) < 0) return -1; if (tDecodeI64(pDecoder, &pRsp->streamId) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->vgId) < 0) return -1; if (tDecodeI32(pDecoder, &pRsp->vgId) < 0) return -1;
if (tDecodeI8(pDecoder, (int8_t*)&pRsp->fillHisFinished) < 0) return -1; if (tDecodeI8(pDecoder, (int8_t *)&pRsp->fillHisFinished) < 0) return -1;
if (tDecodeI64(pDecoder, &pRsp->progressDelay) < 0) return -1; if (tDecodeI64(pDecoder, &pRsp->progressDelay) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->fetchIdx) < 0) return -1; if (tDecodeI32(pDecoder, &pRsp->fetchIdx) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->subFetchIdx) < 0) return -1; if (tDecodeI32(pDecoder, &pRsp->subFetchIdx) < 0) return -1;
return 0; return 0;
} }
int32_t tDeserializeSStreamProgressRsp(void* buf, int32_t bufLen, SStreamProgressRsp* pRsp) { int32_t tDeserializeSStreamProgressRsp(void *buf, int32_t bufLen, SStreamProgressRsp *pRsp) {
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
@ -10438,22 +10438,22 @@ int32_t tDeserializeSStreamProgressRsp(void* buf, int32_t bufLen, SStreamProgres
} }
int32_t tEncodeSMDropTbReqOnSingleVg(SEncoder *pEncoder, const SMDropTbReqsOnSingleVg *pReq) { int32_t tEncodeSMDropTbReqOnSingleVg(SEncoder *pEncoder, const SMDropTbReqsOnSingleVg *pReq) {
const SVgroupInfo* pVgInfo = &pReq->vgInfo; const SVgroupInfo *pVgInfo = &pReq->vgInfo;
if (tEncodeI32(pEncoder, pVgInfo->vgId) < 0) return -1; if (tEncodeI32(pEncoder, pVgInfo->vgId) < 0) return -1;
if (tEncodeU32(pEncoder, pVgInfo->hashBegin) < 0) return -1; if (tEncodeU32(pEncoder, pVgInfo->hashBegin) < 0) return -1;
if (tEncodeU32(pEncoder, pVgInfo->hashEnd) < 0) return -1; if (tEncodeU32(pEncoder, pVgInfo->hashEnd) < 0) return -1;
if (tEncodeSEpSet(pEncoder, &pVgInfo->epSet) < 0) return -1; if (tEncodeSEpSet(pEncoder, &pVgInfo->epSet) < 0) return -1;
if (tEncodeI32(pEncoder, pVgInfo->numOfTable) < 0) return -1; if (tEncodeI32(pEncoder, pVgInfo->numOfTable) < 0) return -1;
int32_t size = pReq->pTbs ? pReq->pTbs->size: 0; int32_t size = pReq->pTbs ? pReq->pTbs->size : 0;
if (tEncodeI32(pEncoder, size) < 0) return -1; if (tEncodeI32(pEncoder, size) < 0) return -1;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
const SVDropTbReq* pInfo = taosArrayGet(pReq->pTbs, i); const SVDropTbReq *pInfo = taosArrayGet(pReq->pTbs, i);
if (tEncodeSVDropTbReq(pEncoder, pInfo) < 0) return -1; if (tEncodeSVDropTbReq(pEncoder, pInfo) < 0) return -1;
} }
return 0; return 0;
} }
int32_t tDecodeSMDropTbReqOnSingleVg(SDecoder* pDecoder, SMDropTbReqsOnSingleVg* pReq) { int32_t tDecodeSMDropTbReqOnSingleVg(SDecoder *pDecoder, SMDropTbReqsOnSingleVg *pReq) {
if (tDecodeI32(pDecoder, &pReq->vgInfo.vgId) < 0) return -1; if (tDecodeI32(pDecoder, &pReq->vgInfo.vgId) < 0) return -1;
if (tDecodeU32(pDecoder, &pReq->vgInfo.hashBegin) < 0) return -1; if (tDecodeU32(pDecoder, &pReq->vgInfo.hashBegin) < 0) return -1;
if (tDecodeU32(pDecoder, &pReq->vgInfo.hashEnd) < 0) return -1; if (tDecodeU32(pDecoder, &pReq->vgInfo.hashEnd) < 0) return -1;
@ -10475,18 +10475,18 @@ int32_t tDecodeSMDropTbReqOnSingleVg(SDecoder* pDecoder, SMDropTbReqsOnSingleVg*
} }
void tFreeSMDropTbReqOnSingleVg(void *p) { void tFreeSMDropTbReqOnSingleVg(void *p) {
SMDropTbReqsOnSingleVg* pReq = p; SMDropTbReqsOnSingleVg *pReq = p;
taosArrayDestroy(pReq->pTbs); taosArrayDestroy(pReq->pTbs);
} }
int32_t tSerializeSMDropTbsReq(void* buf, int32_t bufLen, const SMDropTbsReq* pReq){ int32_t tSerializeSMDropTbsReq(void *buf, int32_t bufLen, const SMDropTbsReq *pReq) {
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
tStartEncode(&encoder); tStartEncode(&encoder);
int32_t size = pReq->pVgReqs ? pReq->pVgReqs->size : 0; int32_t size = pReq->pVgReqs ? pReq->pVgReqs->size : 0;
if (tEncodeI32(&encoder, size) < 0) return -1; if (tEncodeI32(&encoder, size) < 0) return -1;
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
SMDropTbReqsOnSingleVg* pVgReq = taosArrayGet(pReq->pVgReqs, i); SMDropTbReqsOnSingleVg *pVgReq = taosArrayGet(pReq->pVgReqs, i);
if (tEncodeSMDropTbReqOnSingleVg(&encoder, pVgReq) < 0) return -1; if (tEncodeSMDropTbReqOnSingleVg(&encoder, pVgReq) < 0) return -1;
} }
tEndEncode(&encoder); tEndEncode(&encoder);
@ -10495,7 +10495,7 @@ int32_t tSerializeSMDropTbsReq(void* buf, int32_t bufLen, const SMDropTbsReq* pR
return tlen; return tlen;
} }
int32_t tDeserializeSMDropTbsReq(void* buf, int32_t bufLen, SMDropTbsReq* pReq) { int32_t tDeserializeSMDropTbsReq(void *buf, int32_t bufLen, SMDropTbsReq *pReq) {
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
tStartDecode(&decoder); tStartDecode(&decoder);
@ -10516,12 +10516,12 @@ int32_t tDeserializeSMDropTbsReq(void* buf, int32_t bufLen, SMDropTbsReq* pReq)
return 0; return 0;
} }
void tFreeSMDropTbsReq(void* p) { void tFreeSMDropTbsReq(void *p) {
SMDropTbsReq* pReq = p; SMDropTbsReq *pReq = p;
taosArrayDestroyEx(pReq->pVgReqs, tFreeSMDropTbReqOnSingleVg); taosArrayDestroyEx(pReq->pVgReqs, tFreeSMDropTbReqOnSingleVg);
} }
int32_t tEncodeVFetchTtlExpiredTbsRsp(SEncoder* pCoder, const SVFetchTtlExpiredTbsRsp* pRsp) { int32_t tEncodeVFetchTtlExpiredTbsRsp(SEncoder *pCoder, const SVFetchTtlExpiredTbsRsp *pRsp) {
if (tEncodeI32(pCoder, pRsp->vgId) < 0) return -1; if (tEncodeI32(pCoder, pRsp->vgId) < 0) return -1;
int32_t size = pRsp->pExpiredTbs ? pRsp->pExpiredTbs->size : 0; int32_t size = pRsp->pExpiredTbs ? pRsp->pExpiredTbs->size : 0;
if (tEncodeI32(pCoder, size) < 0) return -1; if (tEncodeI32(pCoder, size) < 0) return -1;
@ -10531,7 +10531,7 @@ int32_t tEncodeVFetchTtlExpiredTbsRsp(SEncoder* pCoder, const SVFetchTtlExpiredT
return 0; return 0;
} }
int32_t tDecodeVFetchTtlExpiredTbsRsp(SDecoder* pCoder, SVFetchTtlExpiredTbsRsp* pRsp) { int32_t tDecodeVFetchTtlExpiredTbsRsp(SDecoder *pCoder, SVFetchTtlExpiredTbsRsp *pRsp) {
if (tDecodeI32(pCoder, &pRsp->vgId) < 0) return -1; if (tDecodeI32(pCoder, &pRsp->vgId) < 0) return -1;
int32_t size = 0; int32_t size = 0;
if (tDecodeI32(pCoder, &size) < 0) return -1; if (tDecodeI32(pCoder, &size) < 0) return -1;
@ -10547,7 +10547,7 @@ int32_t tDecodeVFetchTtlExpiredTbsRsp(SDecoder* pCoder, SVFetchTtlExpiredTbsRsp*
return 0; return 0;
} }
void tFreeFetchTtlExpiredTbsRsp(void* p) { void tFreeFetchTtlExpiredTbsRsp(void *p) {
SVFetchTtlExpiredTbsRsp* pRsp = p; SVFetchTtlExpiredTbsRsp *pRsp = p;
taosArrayDestroy(pRsp->pExpiredTbs); taosArrayDestroy(pRsp->pExpiredTbs);
} }