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

feat(tmq): check alter

This commit is contained in:
Liu Jicong 2022-07-22 17:10:40 +08:00
parent 0a47ebc580
commit 2fddb8a6d6
4 changed files with 42 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -480,6 +480,7 @@ typedef struct {
// forbid condition // forbid condition
int64_t ntbUid; int64_t ntbUid;
SArray* ntbColIds; SArray* ntbColIds;
int64_t ctbStbUid;
} SMqTopicObj; } SMqTopicObj;
typedef struct { typedef struct {

49
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -75,7 +75,6 @@ const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]) {
int32_t mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, col_id_t colId) { int32_t mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, col_id_t colId) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL; void *pIter = NULL;
bool found = false;
while (1) { while (1) {
SMqTopicObj *pTopic = NULL; SMqTopicObj *pTopic = NULL;
pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic); pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
@ -96,10 +95,12 @@ int32_t mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, col_id_t colId
SNode *pNode = NULL; SNode *pNode = NULL;
FOREACH(pNode, pNodeList) { FOREACH(pNode, pNodeList) {
SColumnNode *pCol = (SColumnNode *)pNode; SColumnNode *pCol = (SColumnNode *)pNode;
if (pCol->tableId != suid) goto NEXT; if (pCol->tableId != suid && pTopic->ctbStbUid != suid) goto NEXT;
if (pCol->colId > 0 && pCol->colId == colId) { if (pCol->colId > 0 && pCol->colId == colId) {
found = true; sdbRelease(pSdb, pTopic);
goto NEXT; nodesDestroyNode(pAst);
terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
return -1;
} }
mTrace("topic:%s, colId:%d is used", pTopic->name, pCol->colId); mTrace("topic:%s, colId:%d is used", pTopic->name, pCol->colId);
} }
@ -107,10 +108,6 @@ int32_t mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, col_id_t colId
NEXT: NEXT:
sdbRelease(pSdb, pTopic); sdbRelease(pSdb, pTopic);
nodesDestroyNode(pAst); nodesDestroyNode(pAst);
if (found) {
terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
return -1;
}
} }
return 0; return 0;
@ -176,6 +173,7 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
SDB_SET_INT16(pRaw, dataPos, colId, TOPIC_ENCODE_OVER); SDB_SET_INT16(pRaw, dataPos, colId, TOPIC_ENCODE_OVER);
} }
} }
SDB_SET_INT64(pRaw, dataPos, pTopic->ctbStbUid, TOPIC_ENCODE_OVER);
SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER); SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER);
SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER); SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER);
@ -284,6 +282,7 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT16(pRaw, dataPos, &colId, TOPIC_DECODE_OVER); SDB_GET_INT16(pRaw, dataPos, &colId, TOPIC_DECODE_OVER);
taosArrayPush(pTopic->ntbColIds, &colId); taosArrayPush(pTopic->ntbColIds, &colId);
} }
SDB_GET_INT64(pRaw, dataPos, &pTopic->ctbStbUid, TOPIC_DECODE_OVER);
SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER); SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER);
@ -371,15 +370,22 @@ static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
return 0; return 0;
} }
static int32_t extractTopicTbInfo(SNode *pAst, int64_t *ntbUid, SArray *colIds) { static int32_t extractTopicTbInfo(SNode *pAst, SMqTopicObj *pTopic) {
SNodeList *pNodeList = NULL; SNodeList *pNodeList = NULL;
nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList); nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList);
int64_t suid = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->suid;
int8_t tableType = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->tableType;
if (tableType == TSDB_CHILD_TABLE) {
pTopic->ctbStbUid = suid;
} else if (tableType == TSDB_NORMAL_TABLE) {
SNode *pNode = NULL; SNode *pNode = NULL;
FOREACH(pNode, pNodeList) { FOREACH(pNode, pNodeList) {
SColumnNode *pCol = (SColumnNode *)pNode; SColumnNode *pCol = (SColumnNode *)pNode;
if (pCol->tableType != TSDB_NORMAL_TABLE) return -1; if (pCol->tableType == TSDB_NORMAL_TABLE) {
*ntbUid = pCol->tableId; pTopic->ntbUid = pCol->tableId;
taosArrayPush(colIds, &pCol->colId); taosArrayPush(pTopic->ntbColIds, &pCol->colId);
}
}
} }
return 0; return 0;
} }
@ -425,16 +431,16 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
} }
int64_t ntbUid; int64_t ntbUid;
SArray *colIds = taosArrayInit(0, sizeof(int16_t)); topicObj.ntbColIds = taosArrayInit(0, sizeof(int16_t));
if (colIds == NULL) { if (topicObj.ntbColIds == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
if (extractTopicTbInfo(pAst, &ntbUid, colIds) < 0) { extractTopicTbInfo(pAst, &topicObj);
taosArrayDestroy(colIds);
} else { if (topicObj.ntbUid == 0) {
topicObj.ntbUid = ntbUid; taosArrayDestroy(topicObj.ntbColIds);
topicObj.ntbColIds = colIds; topicObj.ntbColIds = NULL;
} }
if (qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema) != 0) { if (qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema) != 0) {
@ -509,6 +515,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
if (code < 0) { if (code < 0) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
mndTransDrop(pTrans); mndTransDrop(pTrans);
ASSERT(0);
return -1; return -1;
} }
void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len); void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
@ -521,7 +528,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
return -1; return -1;
} }
tEncoderClear(&encoder); tEncoderClear(&encoder);
((SMsgHead *)buf)->vgId = pVgroup->vgId; ((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId);
// add redo action // add redo action
STransAction action = {0}; STransAction action = {0};
action.epSet = mndGetVgroupEpset(pMnode, pVgroup); action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
@ -546,7 +553,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
taosMemoryFreeClear(topicObj.physicalPlan); taosMemoryFreeClear(topicObj.physicalPlan);
mndTransDrop(pTrans); mndTransDrop(pTrans);
return 0; return TSDB_CODE_ACTION_IN_PROGRESS;
} }
static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) { static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {

7
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -2667,6 +2667,7 @@ static int32_t jsonToExprNode(const SJson* pJson, void* pObj) {
} }
static const char* jkColumnTableId = "TableId"; static const char* jkColumnTableId = "TableId";
static const char* jkColumnTableType = "TableType";
static const char* jkColumnColId = "ColId"; static const char* jkColumnColId = "ColId";
static const char* jkColumnColType = "ColType"; static const char* jkColumnColType = "ColType";
static const char* jkColumnDbName = "DbName"; static const char* jkColumnDbName = "DbName";
@ -2683,6 +2684,9 @@ static int32_t columnNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkColumnTableId, pNode->tableId); code = tjsonAddIntegerToObject(pJson, jkColumnTableId, pNode->tableId);
} }
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkColumnTableType, pNode->tableType);
}
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkColumnColId, pNode->colId); code = tjsonAddIntegerToObject(pJson, jkColumnColId, pNode->colId);
} }
@ -2718,6 +2722,9 @@ static int32_t jsonToColumnNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetUBigIntValue(pJson, jkColumnTableId, &pNode->tableId); code = tjsonGetUBigIntValue(pJson, jkColumnTableId, &pNode->tableId);
} }
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkColumnTableType, &pNode->tableType);
}
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetSmallIntValue(pJson, jkColumnColId, &pNode->colId); code = tjsonGetSmallIntValue(pJson, jkColumnColId, &pNode->colId);
} }

6
tests/system-test/7-tmq/schema.py
View File

@ -830,9 +830,9 @@ class TDTestCase:
cfgPath = buildPath + "/../sim/psim/cfg" cfgPath = buildPath + "/../sim/psim/cfg"
tdLog.info("cfgPath: %s" % cfgPath) tdLog.info("cfgPath: %s" % cfgPath)
self.tmqCase1(cfgPath, buildPath) # self.tmqCase1(cfgPath, buildPath)
self.tmqCase2(cfgPath, buildPath) # self.tmqCase2(cfgPath, buildPath)
self.tmqCase3(cfgPath, buildPath) # self.tmqCase3(cfgPath, buildPath)
self.tmqCase4(cfgPath, buildPath) self.tmqCase4(cfgPath, buildPath)
self.tmqCase5(cfgPath, buildPath) self.tmqCase5(cfgPath, buildPath)