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

add bSma/rSma param

This commit is contained in:
Cary Xu 2022-03-23 07:58:07 +08:00
parent 273faa1065
commit d4cc6c1fe7
4 changed files with 166 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
include/common/tmsg.h
View File

@ -1363,28 +1363,48 @@ typedef struct {
int64_t tuid; int64_t tuid;
} SDDropTopicReq; } SDDropTopicReq;
typedef struct {
float xFilesFactor;
int8_t delayUnit;
int8_t nFuncIds;
int32_t* pFuncIds;
int64_t delay;
} SRSmaParam;
typedef struct SVCreateTbReq { typedef struct SVCreateTbReq {
int64_t ver; // use a general definition int64_t ver; // use a general definition
char* dbFName; char* dbFName;
char* name; char* name;
uint32_t ttl; uint32_t ttl;
uint32_t keep; uint32_t keep;
uint8_t type; union {
uint8_t info;
struct {
uint8_t rollup : 1; // 1 means rollup sma
uint8_t type : 7;
};
};
union { union {
struct { struct {
tb_uid_t suid; tb_uid_t suid;
uint32_t nCols; uint32_t nCols;
SSchema* pSchema; SSchema* pSchema;
uint32_t nTagCols; uint32_t nTagCols;
SSchema* pTagSchema; SSchema* pTagSchema;
col_id_t nBSmaCols;
col_id_t* pBSmaCols;
SRSmaParam* pRSmaParam;
} stbCfg; } stbCfg;
struct { struct {
tb_uid_t suid; tb_uid_t suid;
SKVRow pTag; SKVRow pTag;
} ctbCfg; } ctbCfg;
struct { struct {
uint32_t nCols; uint32_t nCols;
SSchema* pSchema; SSchema* pSchema;
col_id_t nBSmaCols;
col_id_t* pBSmaCols;
SRSmaParam* pRSmaParam;
} ntbCfg; } ntbCfg;
}; };
} SVCreateTbReq, SVUpdateTbReq; } SVCreateTbReq, SVUpdateTbReq;

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

@ -290,7 +290,7 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
tlen += taosEncodeString(buf, pReq->name); tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeFixedU32(buf, pReq->ttl); tlen += taosEncodeFixedU32(buf, pReq->ttl);
tlen += taosEncodeFixedU32(buf, pReq->keep); tlen += taosEncodeFixedU32(buf, pReq->keep);
tlen += taosEncodeFixedU8(buf, pReq->type); tlen += taosEncodeFixedU8(buf, pReq->info);
switch (pReq->type) { switch (pReq->type) {
case TD_SUPER_TABLE: case TD_SUPER_TABLE:
@ -309,6 +309,20 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes); tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name); tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
} }
tlen += taosEncodeFixedI16(buf, pReq->stbCfg.nBSmaCols);
for (col_id_t i = 0; i < pReq->stbCfg.nBSmaCols; ++i) {
tlen += taosEncodeFixedI16(buf, pReq->stbCfg.pBSmaCols[i]);
}
if(pReq->rollup && NULL != pReq->stbCfg.pRSmaParam) {
SRSmaParam *param = pReq->stbCfg.pRSmaParam;
tlen += taosEncodeFixedU32(buf, (uint32_t)param->xFilesFactor);
tlen += taosEncodeFixedI8(buf, param->delayUnit);
tlen += taosEncodeFixedI8(buf, param->nFuncIds);
for(int8_t i=0; i< param->nFuncIds; ++i) {
tlen += taosEncodeFixedI32(buf, param->pFuncIds[i]);
}
tlen += taosEncodeFixedI64(buf, param->delay);
}
break; break;
case TD_CHILD_TABLE: case TD_CHILD_TABLE:
tlen += taosEncodeFixedI64(buf, pReq->ctbCfg.suid); tlen += taosEncodeFixedI64(buf, pReq->ctbCfg.suid);
@ -322,6 +336,20 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes); tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name); tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name);
} }
tlen += taosEncodeFixedI16(buf, pReq->stbCfg.nBSmaCols);
for (col_id_t i = 0; i < pReq->stbCfg.nBSmaCols; ++i) {
tlen += taosEncodeFixedI16(buf, pReq->stbCfg.pBSmaCols[i]);
}
if(pReq->rollup && NULL != pReq->stbCfg.pRSmaParam) {
SRSmaParam *param = pReq->stbCfg.pRSmaParam;
tlen += taosEncodeFixedU32(buf, (uint32_t)param->xFilesFactor);
tlen += taosEncodeFixedI8(buf, param->delayUnit);
tlen += taosEncodeFixedI8(buf, param->nFuncIds);
for(int8_t i=0; i< param->nFuncIds; ++i) {
tlen += taosEncodeFixedI32(buf, param->pFuncIds[i]);
}
tlen += taosEncodeFixedI64(buf, param->delay);
}
break; break;
default: default:
ASSERT(0); ASSERT(0);
@ -335,7 +363,7 @@ void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
buf = taosDecodeString(buf, &(pReq->name)); buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl)); buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep)); buf = taosDecodeFixedU32(buf, &(pReq->keep));
buf = taosDecodeFixedU8(buf, &(pReq->type)); buf = taosDecodeFixedU8(buf, &(pReq->info));
switch (pReq->type) { switch (pReq->type) {
case TD_SUPER_TABLE: case TD_SUPER_TABLE:
@ -356,6 +384,32 @@ void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes); buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name); buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
} }
buf = taosDecodeFixedI16(buf, &(pReq->stbCfg.nBSmaCols));
if(pReq->stbCfg.nBSmaCols > 0) {
pReq->stbCfg.pBSmaCols = (col_id_t *)malloc(pReq->stbCfg.nBSmaCols * sizeof(col_id_t));
for (col_id_t i = 0; i < pReq->stbCfg.nBSmaCols; ++i) {
buf = taosDecodeFixedI16(buf, pReq->stbCfg.pBSmaCols + i);
}
} else {
pReq->stbCfg.pBSmaCols = NULL;
}
if(pReq->rollup) {
pReq->stbCfg.pRSmaParam = (SRSmaParam *)malloc(sizeof(SRSmaParam));
SRSmaParam *param = pReq->stbCfg.pRSmaParam;
buf = taosDecodeFixedU32(buf, (uint32_t*)&param->xFilesFactor);
buf = taosDecodeFixedI8(buf, &param->delayUnit);
buf = taosDecodeFixedI8(buf, &param->nFuncIds);
if(param->nFuncIds > 0) {
for (int8_t i = 0; i< param->nFuncIds; ++i) {
buf = taosDecodeFixedI32(buf, param->pFuncIds + i);
}
} else {
param->pFuncIds = NULL;
}
buf = taosDecodeFixedI64(buf, &param->delay);
} else {
pReq->stbCfg.pRSmaParam = NULL;
}
break; break;
case TD_CHILD_TABLE: case TD_CHILD_TABLE:
buf = taosDecodeFixedI64(buf, &pReq->ctbCfg.suid); buf = taosDecodeFixedI64(buf, &pReq->ctbCfg.suid);
@ -370,6 +424,32 @@ void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes); buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name); buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name);
} }
buf = taosDecodeFixedI16(buf, &(pReq->stbCfg.nBSmaCols));
if(pReq->stbCfg.nBSmaCols > 0) {
pReq->stbCfg.pBSmaCols = (col_id_t *)malloc(pReq->stbCfg.nBSmaCols * sizeof(col_id_t));
for (col_id_t i = 0; i < pReq->stbCfg.nBSmaCols; ++i) {
buf = taosDecodeFixedI16(buf, pReq->stbCfg.pBSmaCols + i);
}
} else {
pReq->stbCfg.pBSmaCols = NULL;
}
if(pReq->rollup) {
pReq->stbCfg.pRSmaParam = (SRSmaParam *)malloc(sizeof(SRSmaParam));
SRSmaParam *param = pReq->stbCfg.pRSmaParam;
buf = taosDecodeFixedU32(buf, (uint32_t*)&param->xFilesFactor);
buf = taosDecodeFixedI8(buf, &param->delayUnit);
buf = taosDecodeFixedI8(buf, &param->nFuncIds);
if(param->nFuncIds > 0) {
for (int8_t i = 0; i< param->nFuncIds; ++i) {
buf = taosDecodeFixedI32(buf, param->pFuncIds + i);
}
} else {
param->pFuncIds = NULL;
}
buf = taosDecodeFixedI64(buf, &param->delay);
} else {
pReq->stbCfg.pRSmaParam = NULL;
}
break; break;
default: default:
ASSERT(0); ASSERT(0);

2
source/dnode/vnode/src/meta/metaBDBImpl.c
View File

@ -527,7 +527,7 @@ static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg) {
buf = taosDecodeString(buf, &(pTbCfg->name)); buf = taosDecodeString(buf, &(pTbCfg->name));
buf = taosDecodeFixedU32(buf, &(pTbCfg->ttl)); buf = taosDecodeFixedU32(buf, &(pTbCfg->ttl));
buf = taosDecodeFixedU32(buf, &(pTbCfg->keep)); buf = taosDecodeFixedU32(buf, &(pTbCfg->keep));
buf = taosDecodeFixedU8(buf, &(pTbCfg->type)); buf = taosDecodeFixedU8(buf, &(pTbCfg->info));
if (pTbCfg->type == META_SUPER_TABLE) { if (pTbCfg->type == META_SUPER_TABLE) {
SSchemaWrapper sw; SSchemaWrapper sw;

55
source/dnode/vnode/test/tsdbSmaTest.cpp
View File

@ -33,6 +33,61 @@ int main(int argc, char **argv) {
return RUN_ALL_TESTS(); return RUN_ALL_TESTS();
} }
TEST(testCase, unionEncodeDecodeTest) {
typedef struct {
union {
uint8_t info;
struct {
uint8_t rollup : 1; // 1 means rollup sma
uint8_t type : 7;
};
};
col_id_t nBSmaCols;
col_id_t* pBSmaCols;
} SUnionTest;
SUnionTest sut = {0};
sut.rollup = 1;
sut.type = 1;
sut.nBSmaCols = 2;
sut.pBSmaCols = (col_id_t*)malloc(sut.nBSmaCols * sizeof(col_id_t));
for (col_id_t i = 0; i < sut.nBSmaCols; ++i) {
sut.pBSmaCols[i] = i + 100;
}
void* buf = malloc(1024);
void * pBuf = buf;
int32_t tlen = 0;
tlen += taosEncodeFixedU8(&buf, sut.info);
tlen += taosEncodeFixedI16(&buf, sut.nBSmaCols);
for (col_id_t i = 0; i < sut.nBSmaCols; ++i) {
tlen += taosEncodeFixedI16(&buf, sut.pBSmaCols[i]);
}
SUnionTest dut = {0};
pBuf = taosDecodeFixedU8(pBuf, &dut.info);
pBuf = taosDecodeFixedI16(pBuf, &dut.nBSmaCols);
if(dut.nBSmaCols > 0) {
dut.pBSmaCols = (col_id_t*)malloc(dut.nBSmaCols * sizeof(col_id_t));
for(col_id_t i=0; i < dut.nBSmaCols; ++i) {
pBuf = taosDecodeFixedI16(pBuf, dut.pBSmaCols + i);
}
} else {
dut.pBSmaCols = NULL;
}
printf("sut.rollup=%" PRIu8 ", type=%" PRIu8 ", info=%" PRIu8 "\n", sut.rollup, sut.type, sut.info);
printf("dut.rollup=%" PRIu8 ", type=%" PRIu8 ", info=%" PRIu8 "\n", dut.rollup, dut.type, dut.info);
ASSERT_EQ(sut.rollup, dut.rollup);
ASSERT_EQ(sut.type, dut.type);
ASSERT_EQ(sut.nBSmaCols, dut.nBSmaCols);
for (col_id_t i = 0; i< sut.nBSmaCols; ++i) {
ASSERT_EQ(*(col_id_t*)(sut.pBSmaCols + i), sut.pBSmaCols[i]);
ASSERT_EQ(*(col_id_t*)(sut.pBSmaCols + i), dut.pBSmaCols[i]);
}
}
#if 1 #if 1
TEST(testCase, tSma_Meta_Encode_Decode_Test) { TEST(testCase, tSma_Meta_Encode_Decode_Test) {
// encode // encode