Skip to content

T
TDengine

taosdata / TDengine
12 个月 前同步成功

通知 1180
Star 22014
Fork 4786
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 ad3de2d9 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #14717 from taosdata/feature/TD-13041 

feat:write meta from tmq to taosd
上级 4e777828 b4cadcbe
隐藏空白更改
内联 并排
Showing with 93 addition and 78 deletion
include/common/tmsg.h
浏览文件 @ ad3de2d9
......@@ -169,6 +169,9 @@ typedef enum _mgmt_table {
#define TD_CHILD_TABLE TSDB_CHILD_TABLE
#define TD_NORMAL_TABLE TSDB_NORMAL_TABLE
#define TD_REQ_FROM_APP 0
#define TD_REQ_FROM_TAOX 1
typedef struct {
int32_t vgId;
char* dbFName;
......@@ -432,30 +435,30 @@ static FORCE_INLINE int32_t tDecodeSSchemaWrapperEx(SDecoder* pDecoder, SSchemaW
STSchema* tdGetSTSChemaFromSSChema(SSchema** pSchema, int32_t nCols);
typedef struct {
char name[TSDB_TABLE_FNAME_LEN];
int8_t igExists;
int64_t delay1;
int64_t delay2;
int64_t watermark1;
int64_t watermark2;
int32_t ttl;
int32_t numOfColumns;
int32_t numOfTags;
int32_t numOfFuncs;
int32_t commentLen;
int32_t ast1Len;
int32_t ast2Len;
SArray* pColumns; // array of SField
int32_t cVersion;
SArray* pTags; // array of SField
int32_t tVersion;
SArray* pFuncs;
char* pComment;
char* pAst1;
char* pAst2;
tb_uid_t suid;
int8_t source; // 1-taosX or 0-taosClient
int8_t reserved[8];
char name[TSDB_TABLE_FNAME_LEN];
int8_t igExists;
int8_t source; // 1-taosX or 0-taosClient
int8_t reserved[6];
tb_uid_t suid;
int64_t delay1;
int64_t delay2;
int64_t watermark1;
int64_t watermark2;
int32_t ttl;
int32_t colVer;
int32_t tagVer;
int32_t numOfColumns;
int32_t numOfTags;
int32_t numOfFuncs;
int32_t commentLen;
int32_t ast1Len;
int32_t ast2Len;
SArray* pColumns; // array of SField
SArray* pTags; // array of SField
SArray* pFuncs;
char* pComment;
char* pAst1;
char* pAst2;
} SMCreateStbReq;
int32_t tSerializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pReq);
......@@ -463,11 +466,11 @@ int32_t tDeserializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pR
void tFreeSMCreateStbReq(SMCreateStbReq* pReq);
typedef struct {
char name[TSDB_TABLE_FNAME_LEN];
int8_t igNotExists;
tb_uid_t suid;
int8_t source; // 1-taosX or 0-taosClient
int8_t reserved[8];
char name[TSDB_TABLE_FNAME_LEN];
int8_t igNotExists;
int8_t source; // 1-taosX or 0-taosClient
int8_t reserved[6];
tb_uid_t suid;
} SMDropStbReq;
int32_t tSerializeSMDropStbReq(void* buf, int32_t bufLen, SMDropStbReq* pReq);
......@@ -476,8 +479,6 @@ int32_t tDeserializeSMDropStbReq(void* buf, int32_t bufLen, SMDropStbReq* pReq);
typedef struct {
char name[TSDB_TABLE_FNAME_LEN];
int8_t alterType;
int32_t tagVer;
int32_t colVer;
int32_t numOfFields;
SArray* pFields;
int32_t ttl;
......
include/util/taoserror.h
浏览文件 @ ad3de2d9
......@@ -247,9 +247,11 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_COLUMN_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x03AE)
#define TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC TAOS_DEF_ERROR_CODE(0, 0x03AF)
#define TSDB_CODE_MND_SINGLE_STB_MODE_DB TAOS_DEF_ERROR_CODE(0, 0x03B0)
#define TSDB_CODE_MND_INVALID_SCHEMA_VER TAOS_DEF_ERROR_CODE(0, 0x03B1)
#define TSDB_CODE_MND_STABLE_UID_NOT_MATCH TAOS_DEF_ERROR_CODE(0, 0x03B2)
// mnode-infoSchema
#define TSDB_CODE_MND_INVALID_SYS_TABLENAME TAOS_DEF_ERROR_CODE(0, 0x03BA)
#define TSDB_CODE_MND_INVALID_SYS_TABLENAME TAOS_DEF_ERROR_CODE(0, 0x03BA)
// mnode-func
#define TSDB_CODE_MND_FUNC_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x03C0)
......
source/client/src/tmq.c
浏览文件 @ ad3de2d9
......@@ -2305,8 +2305,8 @@ static int32_t taosCreateStb(TAOS *taos, void *meta, int32_t metaLen){
strcpy(field.name, pSchema->name);
taosArrayPush(pReq.pTags, &field);
}
pReq.cVersion = req.schemaRow.version;
pReq.tVersion = req.schemaTag.version;
pReq.colVer = req.schemaRow.version;
pReq.tagVer = req.schemaTag.version;
pReq.numOfColumns = req.schemaRow.nCols;
pReq.numOfTags = req.schemaTag.nCols;
pReq.commentLen = -1;
......
source/common/src/tmsg.c
浏览文件 @ ad3de2d9
......@@ -496,11 +496,18 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
if (tEncodeI8(&encoder, pReq->igExists) < 0) return -1;
if (tEncodeI8(&encoder, pReq->source) < 0) return -1;
for (int32_t i = 0; i < sizeof(pReq->reserved) / sizeof(int8_t); ++i) {
if (tEncodeI8(&encoder, pReq->reserved[i]) < 0) return -1;
}
if (tEncodeI64(&encoder, pReq->suid) < 0) return -1;
if (tEncodeI64(&encoder, pReq->delay1) < 0) return -1;
if (tEncodeI64(&encoder, pReq->delay2) < 0) return -1;
if (tEncodeI64(&encoder, pReq->watermark1) < 0) return -1;
if (tEncodeI64(&encoder, pReq->watermark2) < 0) return -1;
if (tEncodeI32(&encoder, pReq->ttl) < 0) return -1;
if (tEncodeI32(&encoder, pReq->colVer) < 0) return -1;
if (tEncodeI32(&encoder, pReq->tagVer) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfColumns) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfTags) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfFuncs) < 0) return -1;
......@@ -516,8 +523,6 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (tEncodeCStr(&encoder, pField->name) < 0) return -1;
}
if (tEncodeI32(&encoder, pReq->cVersion) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfTags; ++i) {
SField *pField = taosArrayGet(pReq->pTags, i);
if (tEncodeI8(&encoder, pField->type) < 0) return -1;
......@@ -526,8 +531,6 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (tEncodeCStr(&encoder, pField->name) < 0) return -1;
}
if (tEncodeI32(&encoder, pReq->tVersion) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfFuncs; ++i) {
const char *pFunc = taosArrayGet(pReq->pFuncs, i);
if (tEncodeCStr(&encoder, pFunc) < 0) return -1;
......@@ -542,11 +545,6 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (pReq->ast2Len > 0) {
if (tEncodeBinary(&encoder, pReq->pAst2, pReq->ast2Len) < 0) return -1;
}
if (tEncodeI64(&encoder, pReq->suid) < 0) return -1;
if (tEncodeI8(&encoder, pReq->source) < 0) return -1;
for (int32_t i = 0; i < sizeof(pReq->reserved)/sizeof(int8_t); ++i) {
if (tEncodeI8(&encoder, pReq->reserved[i]) < 0) return -1;
}
tEndEncode(&encoder);
......@@ -562,11 +560,18 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->igExists) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->source) < 0) return -1;
for (int32_t i = 0; i < sizeof(pReq->reserved) / sizeof(int8_t); ++i) {
if (tDecodeI8(&decoder, &pReq->reserved[i]) < 0) return -1;
}
if (tDecodeI64(&decoder, &pReq->suid) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->delay1) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->delay2) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->watermark1) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->watermark2) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->ttl) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->colVer) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->tagVer) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfColumns) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfTags) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfFuncs) < 0) return -1;
......@@ -594,8 +599,6 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
}
}
if (tDecodeI32(&decoder, &pReq->cVersion) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfTags; ++i) {
SField field = {0};
if (tDecodeI8(&decoder, &field.type) < 0) return -1;
......@@ -608,8 +611,6 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
}
}
if (tDecodeI32(&decoder, &pReq->tVersion) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfFuncs; ++i) {
char pFunc[TSDB_FUNC_NAME_LEN] = {0};
if (tDecodeCStrTo(&decoder, pFunc) < 0) return -1;
......@@ -637,12 +638,6 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (tDecodeCStrTo(&decoder, pReq->pAst2) < 0) return -1;
}
if (tDecodeI64(&decoder, &pReq->suid) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->source) < 0) return -1;
for (int32_t i = 0; i < sizeof(pReq->reserved)/sizeof(int8_t); ++i) {
if (tDecodeI8(&decoder, &pReq->reserved[i]) < 0) return -1;
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
......@@ -664,11 +659,11 @@ int32_t tSerializeSMDropStbReq(void *buf, int32_t bufLen, SMDropStbReq *pReq) {
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
if (tEncodeI8(&encoder, pReq->igNotExists) < 0) return -1;
if (tEncodeI64(&encoder, pReq->suid) < 0) return -1;
if (tEncodeI8(&encoder, pReq->source) < 0) return -1;
for (int32_t i = 0; i < sizeof(pReq->reserved)/sizeof(int8_t); ++i) {
for (int32_t i = 0; i < sizeof(pReq->reserved) / sizeof(int8_t); ++i) {
if (tEncodeI8(&encoder, pReq->reserved[i]) < 0) return -1;
}
if (tEncodeI64(&encoder, pReq->suid) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -683,11 +678,11 @@ int32_t tDeserializeSMDropStbReq(void *buf, int32_t bufLen, SMDropStbReq *pReq)
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->igNotExists) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->suid) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->source) < 0) return -1;
for (int32_t i = 0; i < sizeof(pReq->reserved)/sizeof(int8_t); ++i) {
for (int32_t i = 0; i < sizeof(pReq->reserved) / sizeof(int8_t); ++i) {
if (tDecodeI8(&decoder, &pReq->reserved[i]) < 0) return -1;
}
if (tDecodeI64(&decoder, &pReq->suid) < 0) return -1;
tEndDecode(&decoder);
......@@ -702,8 +697,6 @@ int32_t tSerializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq)
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
if (tEncodeI8(&encoder, pReq->alterType) < 0) return -1;
if (tEncodeI32(&encoder, pReq->tagVer) < 0) return -1;
if (tEncodeI32(&encoder, pReq->colVer) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfFields) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfFields; ++i) {
SField *pField = taosArrayGet(pReq->pFields, i);
......@@ -730,8 +723,6 @@ int32_t tDeserializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->alterType) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->tagVer) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->colVer) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfFields) < 0) return -1;
pReq->pFields = taosArrayInit(pReq->numOfFields, sizeof(SField));
if (pReq->pFields == NULL) {
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ ad3de2d9
......@@ -707,8 +707,8 @@ int32_t mndBuildStbFromReq(SMnode *pMnode, SStbObj *pDst, SMCreateStbReq *pCreat
pDst->updateTime = pDst->createdTime;
pDst->uid = (pCreate->source == 1) ? pCreate->suid : mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
pDst->dbUid = pDb->uid;
pDst->tagVer = (pCreate->source == 1) ? pCreate->tVersion : 1;
pDst->colVer = (pCreate->source == 1) ? pCreate->cVersion : 1;
pDst->tagVer = (pCreate->source != TD_REQ_FROM_APP) ? pCreate->tagVer : 1;
pDst->colVer = (pCreate->source != TD_REQ_FROM_APP) ? pCreate->colVer : 1;
pDst->smaVer = 1;
pDst->nextColId = 1;
pDst->maxdelay[0] = pCreate->delay1;
......@@ -869,9 +869,38 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
pStb = mndAcquireStb(pMnode, createReq.name);
if (pStb != NULL) {
if (createReq.igExists) {
mDebug("stb:%s, already exist, ignore exist is set", createReq.name);
code = 0;
goto _OVER;
if (createReq.source == TD_REQ_FROM_APP) {
mDebug("stb:%s, already exist, ignore exist is set", createReq.name);
code = 0;
goto _OVER;
} else if (pStb->uid != createReq.suid) {
mError("stb:%s, already exist while create, input suid:%" PRId64 " not match with exist suid:%" PRId64,
createReq.name, createReq.suid, pStb->uid);
terrno = TSDB_CODE_MND_STABLE_UID_NOT_MATCH;
goto _OVER;
} else if (createReq.tagVer > 0 || createReq.colVer > 0) {
int32_t tagDelta = pStb->tagVer - createReq.tagVer;
int32_t colDelta = pStb->colVer - createReq.colVer;
int32_t verDelta = tagDelta + verDelta;
mInfo("stb:%s, already exist while create, input tagVer:%d colVer:%d, exist tagVer:%d colVer:%d",
createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
if (tagDelta <= 0 && colDelta <= 0) {
mInfo("stb:%s, schema version is not incremented and nothing needs to be done", createReq.name);
code = 0;
goto _OVER;
} else if ((tagDelta == 1 || colDelta == 1) && (verDelta == 1)) {
mInfo("stb:%s, schema version is only increased by 1 number, do alter operation", createReq.name);
} else {
mError("stb:%s, schema version increase more than 1 number, error is returned", createReq.name);
terrno = TSDB_CODE_MND_INVALID_SCHEMA_VER;
goto _OVER;
}
} else {
mError("stb:%s, already exist while create, input tagVer:%d colVer:%d is invalid", createReq.name,
createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
terrno = TSDB_CODE_MND_INVALID_SCHEMA_VER;
goto _OVER;
}
} else {
terrno = TSDB_CODE_MND_STB_ALREADY_EXIST;
goto _OVER;
......@@ -1614,14 +1643,6 @@ static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {
goto _OVER;
}
if ((alterReq.tagVer > 0 && alterReq.colVer > 0) &&
(alterReq.tagVer <= pStb->tagVer || alterReq.colVer <= pStb->colVer)) {
mDebug("stb:%s, already exist, tagVer:%d colVer:%d smaller than in mnode, tagVer:%d colVer:%d, alter success",
alterReq.name, alterReq.tagVer, alterReq.colVer, pStb->tagVer, pStb->colVer);
code = 0;
goto _OVER;
}
if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb) != 0) {
goto _OVER;
}
......@@ -1752,7 +1773,7 @@ static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
}
}
if (dropReq.source == 1 && pStb->uid != dropReq.suid){
if (dropReq.source != TD_REQ_FROM_APP && pStb->uid != dropReq.suid) {
terrno = TSDB_CODE_MND_STB_NOT_EXIST;
goto _OVER;
}
......
source/dnode/mnode/impl/test/stb/stb.cpp
浏览文件 @ ad3de2d9
......@@ -277,8 +277,6 @@ void* MndTestStb::BuildAlterStbUpdateColumnBytesReq(const char* stbname, const c
req.numOfFields = 1;
req.pFields = taosArrayInit(1, sizeof(SField));
req.alterType = TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES;
req.tagVer = verInBlock;
req.colVer = verInBlock;
SField field = {0};
field.bytes = bytes;
......@@ -818,7 +816,7 @@ TEST_F(MndTestStb, 08_Alter_Stb_AlterTagBytes) {
{
void* pReq = BuildAlterStbUpdateColumnBytesReq(stbname, "col_not_exist", 20, &contLen, 1);
SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_STB, pReq, contLen);
ASSERT_EQ(pRsp->code, 0);
ASSERT_EQ(pRsp->code, TSDB_CODE_MND_COLUMN_NOT_EXIST);
test.SendShowReq(TSDB_MGMT_TABLE_STB, "user_stables", dbname);
EXPECT_EQ(test.GetShowRows(), 1);
......
source/util/src/terror.c
浏览文件 @ ad3de2d9
......@@ -251,6 +251,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_COLUMN_ALREADY_EXIST, "Column already exists
TAOS_DEFINE_ERROR(TSDB_CODE_MND_COLUMN_NOT_EXIST, "Column does not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC,"Field used by topic")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_SINGLE_STB_MODE_DB, "Database is single stable mode")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SCHEMA_VER, "Invalid schema version while alter stb")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_STABLE_UID_NOT_MATCH, "Invalid stable uid while alter stb")
// mnode-infoSchema
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SYS_TABLENAME, "Invalid system table name")
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或