Merge pull request #15320 from taosdata/fix/mnode

fix: modifying stb in subscription is not allowed

source/dnode/mnode/impl/inc/mndTopic.h

@@ -37,7 +37,7 @@ const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]);
 
 int32_t mndSetTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic);
 
-int32_t mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, col_id_t colId);
+int32_t mndCheckColAndTagModifiable(SMnode *pMnode,  const char* stbname, int64_t suid, col_id_t colId);
 
 #ifdef __cplusplus
 }

source/dnode/mnode/impl/src/mndStb.c

@@ -1144,7 +1144,7 @@ static int32_t mndDropSuperTableTag(SMnode *pMnode, const SStbObj *pOld, SStbObj
   }
 
   col_id_t colId = pOld->pTags[tag].colId;
-  if (mndCheckColAndTagModifiable(pMnode, pOld->uid, colId) != 0) {
+  if (mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId) != 0) {
     return -1;
   }
 
@@ -1179,7 +1179,7 @@ static int32_t mndAlterStbTagName(SMnode *pMnode, const SStbObj *pOld, SStbObj *
   }
 
   col_id_t colId = pOld->pTags[tag].colId;
-  if (mndCheckColAndTagModifiable(pMnode, pOld->uid, colId) != 0) {
+  if (mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId) != 0) {
     return -1;
   }
 
@@ -1213,7 +1213,7 @@ static int32_t mndAlterStbTagBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj
   }
 
   col_id_t colId = pOld->pTags[tag].colId;
-  if (mndCheckColAndTagModifiable(pMnode, pOld->uid, colId) != 0) {
+  if (mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId) != 0) {
     return -1;
   }
 
@@ -1295,7 +1295,7 @@ static int32_t mndDropSuperTableColumn(SMnode *pMnode, const SStbObj *pOld, SStb
   }
 
   col_id_t colId = pOld->pColumns[col].colId;
-  if (mndCheckColAndTagModifiable(pMnode, pOld->uid, colId) != 0) {
+  if (mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId) != 0) {
     return -1;
   }
 
@@ -1329,7 +1329,7 @@ static int32_t mndAlterStbColumnBytes(SMnode *pMnode, const SStbObj *pOld, SStbO
   }
 
   col_id_t colId = pOld->pColumns[col].colId;
-  if (mndCheckColAndTagModifiable(pMnode, pOld->uid, colId) != 0) {
+  if (mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId) != 0) {
     return -1;
   }
 

source/dnode/mnode/impl/src/mndTopic.c

@@ -72,13 +72,16 @@ const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]) {
   return strchr(topic, '.') + 1;
 }
 
-int32_t mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, col_id_t colId) {
+int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char* stbname, int64_t suid, col_id_t colId) {
   SSdb *pSdb = pMnode->pSdb;
   void *pIter = NULL;
   while (1) {
     SMqTopicObj *pTopic = NULL;
     pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
     if (pIter == NULL) break;
+
+    mDebug("topic:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, subType:%d sql:%s", 
+          pTopic->name, stbname, suid, colId, pTopic->subType, pTopic->sql);
     if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) {
       sdbRelease(pSdb, pTopic);
       continue;
@@ -95,14 +98,20 @@ int32_t mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, col_id_t colId
     SNode *pNode = NULL;
     FOREACH(pNode, pNodeList) {
       SColumnNode *pCol = (SColumnNode *)pNode;
-      if (pCol->tableId != suid && pTopic->ctbStbUid != suid) goto NEXT;
+      mDebug("topic:%s, check colId:%d tableId:%" PRId64 " ctbStbUid:%" PRId64, pTopic->name, pCol->colId, pCol->tableId, pTopic->ctbStbUid);
+   
+      if (pCol->tableId != suid && pTopic->ctbStbUid != suid) {
+        mDebug("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
+        goto NEXT;
+      }
       if (pCol->colId > 0 && pCol->colId == colId) {
         sdbRelease(pSdb, pTopic);
         nodesDestroyNode(pAst);
         terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
+        mError("topic:%s, check colId:%d conflicted", pTopic->name, pCol->colId);
         return -1;
       }
-      mTrace("topic:%s, colId:%d is used", pTopic->name, pCol->colId);
+      mDebug("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
     }
 
   NEXT: