Merge pull request #10166 from taosdata/feature/mnode

update stb

include/common/tmsg.h

@@ -111,8 +111,8 @@ typedef enum _mgmt_table {
   TSDB_MGMT_TABLE_MAX,
 } EShowType;
 
-#define TSDB_ALTER_TABLE_ADD_TAG_COLUMN 1
-#define TSDB_ALTER_TABLE_DROP_TAG_COLUMN 2
+#define TSDB_ALTER_TABLE_ADD_TAG 1
+#define TSDB_ALTER_TABLE_DROP_TAG 2
 #define TSDB_ALTER_TABLE_UPDATE_TAG_NAME 3
 #define TSDB_ALTER_TABLE_UPDATE_TAG_VAL 4
 
@@ -1199,8 +1199,8 @@ typedef struct {
 
 int32_t tSerializeSVCreateTbBatchReq(void** buf, SVCreateTbBatchReq* pReq);
 void*   tDeserializeSVCreateTbBatchReq(void* buf, SVCreateTbBatchReq* pReq);
-int32_t tSerializeSVCreateTbBatchReqp(void** buf, SVCreateTbBatchReq* pRsp);
-void*   tDeserializeSVCreateTbBatchReq(void* buf, SVCreateTbBatchReq* pRsp);
+int32_t tSerializeSVCreateTbBatchRsp(void** buf, SVCreateTbBatchRsp* pRsp);
+void*   tDeserializeSVCreateTbBatchRsp(void* buf, SVCreateTbBatchRsp* pRsp);
 
 typedef struct {
   uint64_t ver;
@@ -1210,7 +1210,6 @@ typedef struct {
 } SVDropTbReq;
 
 typedef struct {
-  uint64_t ver;
 } SVDropTbRsp;
 
 int32_t tSerializeSVDropTbReq(void** buf, SVDropTbReq* pReq);

include/libs/sync/sync.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <cli@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_LIBS_SYNC_H
+#define _TD_LIBS_SYNC_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+#include "taosdef.h"
+
+typedef int32_t  SyncNodeId;
+typedef int32_t  SyncGroupId;
+typedef int64_t  SyncIndex;
+typedef uint64_t SyncTerm;
+
+typedef enum {
+  TAOS_SYNC_STATE_FOLLOWER = 0,
+  TAOS_SYNC_STATE_CANDIDATE = 1,
+  TAOS_SYNC_STATE_LEADER = 2,
+} ESyncState;
+
+typedef struct {
+  void*  data;
+  size_t len;
+} SSyncBuffer;
+
+typedef struct {
+  SyncNodeId nodeId;
+  uint16_t   nodePort;                 // node sync Port
+  char       nodeFqdn[TSDB_FQDN_LEN];  // node FQDN
+} SNodeInfo;
+
+typedef struct {
+  int32_t   selfIndex;
+  int32_t   replica;
+  SNodeInfo nodeInfo[TSDB_MAX_REPLICA];
+} SSyncCluster;
+
+typedef struct {
+  int32_t   selfIndex;
+  int32_t   replica;
+  SNodeInfo node[TSDB_MAX_REPLICA];
+  ESyncState role[TSDB_MAX_REPLICA];
+} SNodesRole;
+
+typedef struct SSyncFSM {
+  void* pData;
+
+  // apply committed log, bufs will be free by sync module
+  int32_t (*applyLog)(struct SSyncFSM* fsm, SyncIndex index, const SSyncBuffer* buf, void* pData);
+
+  // cluster commit callback
+  int32_t (*onClusterChanged)(struct SSyncFSM* fsm, const SSyncCluster* cluster, void* pData);
+
+  // fsm return snapshot in ppBuf, bufs will be free by sync module
+  // TODO: getSnapshot SHOULD be async?
+  int32_t (*getSnapshot)(struct SSyncFSM* fsm, SSyncBuffer** ppBuf, int32_t* objId, bool* isLast);
+
+  // fsm apply snapshot with pBuf data
+  int32_t (*applySnapshot)(struct SSyncFSM* fsm, SSyncBuffer* pBuf, int32_t objId, bool isLast);
+
+  // call when restore snapshot and log done
+  int32_t (*onRestoreDone)(struct SSyncFSM* fsm);
+
+  void (*onRollback)(struct SSyncFSM* fsm, SyncIndex index, const SSyncBuffer* buf);
+
+  void (*onRoleChanged)(struct SSyncFSM* fsm, const SNodesRole* pRole);
+
+} SSyncFSM;
+
+typedef struct SSyncLogStore {
+  void* pData;
+
+  // write log with given index
+  int32_t (*logWrite)(struct SSyncLogStore* logStore, SyncIndex index, SSyncBuffer* pBuf);
+
+  /** 
+   * read log from given index(included) with limit, return the actual num in nBuf,
+   * pBuf will be free in sync module
+   **/
+  int32_t (*logRead)(struct SSyncLogStore* logStore, SyncIndex index, int limit,
+                      SSyncBuffer* pBuf, int* nBuf);
+
+  // mark log with given index has been commtted
+  int32_t (*logCommit)(struct SSyncLogStore* logStore, SyncIndex index);
+
+  // prune log before given index(not included)
+  int32_t (*logPrune)(struct SSyncLogStore* logStore, SyncIndex index);
+
+  // rollback log after given index(included)
+  int32_t (*logRollback)(struct SSyncLogStore* logStore, SyncIndex index);
+
+  // return last index of log
+  SyncIndex (*logLastIndex)(struct SSyncLogStore* logStore);
+} SSyncLogStore;
+
+typedef struct SStateManager {
+  void* pData;
+
+  // save serialized server state data, buffer will be free by Sync
+  int32_t (*saveServerState)(struct SStateManager* stateMng, const char* buffer, int n);
+
+  // read serialized server state data, buffer will be free by Sync
+  int32_t (*readServerState)(struct SStateManager* stateMng, char** ppBuffer, int* n);
+
+  // save serialized cluster state data, buffer will be free by Sync
+  void (*saveClusterState)(struct SStateManager* stateMng, const char* buffer, int n);
+
+  // read serialized cluster state data, buffer will be free by Sync
+  int32_t (*readClusterState)(struct SStateManager* stateMng, char** ppBuffer, int* n);
+} SStateManager;
+
+typedef struct {
+  SyncGroupId   vgId;
+  SyncIndex     appliedIndex;
+  SSyncCluster  syncCfg;
+  SSyncFSM      fsm;
+  SSyncLogStore logStore;
+  SStateManager stateManager;
+} SSyncInfo;
+
+struct SSyncNode;
+typedef struct SSyncNode SSyncNode;
+
+int32_t syncInit();
+void    syncCleanUp();
+
+SSyncNode* syncStart(const SSyncInfo*);
+void       syncReconfig(const SSyncNode*, const SSyncCluster*);
+void       syncStop(const SSyncNode*);
+
+int32_t syncPropose(SSyncNode* syncNode, const SSyncBuffer* pBuf, void* pData, bool isWeak);
+
+int32_t syncAddNode(SSyncNode syncNode, const SNodeInfo *pNode);
+
+int32_t syncRemoveNode(SSyncNode syncNode, const SNodeInfo *pNode);
+
+extern int32_t sDebugFlag;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_LIBS_SYNC_H*/

include/util/taoserror.h

@@ -218,18 +218,19 @@ int32_t* taosGetErrno();
 // mnode-stable
 #define TSDB_CODE_MND_STB_ALREADY_EXIST         TAOS_DEF_ERROR_CODE(0, 0x03A0)
 #define TSDB_CODE_MND_STB_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x03A1)
-#define TSDB_CODE_MND_TOO_MANY_STBS             TAOS_DEF_ERROR_CODE(0, 0x03A2)
-#define TSDB_CODE_MND_INVALID_STB               TAOS_DEF_ERROR_CODE(0, 0x03A3)
-#define TSDB_CODE_MND_INVALID_STB_OPTION        TAOS_DEF_ERROR_CODE(0, 0x03A4)
-#define TSDB_CODE_MND_STB_OPTION_UNCHNAGED      TAOS_DEF_ERROR_CODE(0, 0x03A5)
-#define TSDB_CODE_MND_TOO_MANY_TAGS             TAOS_DEF_ERROR_CODE(0, 0x03A6)
-#define TSDB_CODE_MND_TAG_ALREAY_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03A7)
-#define TSDB_CODE_MND_TAG_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x03A8)
-#define TSDB_CODE_MND_TOO_MANY_COLUMNS          TAOS_DEF_ERROR_CODE(0, 0x03A9)
-#define TSDB_CODE_MND_COLUMN_ALREAY_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03AA)
-#define TSDB_CODE_MND_COLUMN_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03AB)
-#define TSDB_CODE_MND_INVALID_ROW_BYTES         TAOS_DEF_ERROR_CODE(0, 0x03AC)
-#define TSDB_CODE_MND_NAME_CONFLICT_WITH_TOPIC  TAOS_DEF_ERROR_CODE(0, 0x03AD)
+#define TSDB_CODE_MND_NAME_CONFLICT_WITH_TOPIC  TAOS_DEF_ERROR_CODE(0, 0x03A2)
+#define TSDB_CODE_MND_TOO_MANY_STBS             TAOS_DEF_ERROR_CODE(0, 0x03A3)
+#define TSDB_CODE_MND_INVALID_STB               TAOS_DEF_ERROR_CODE(0, 0x03A4)
+#define TSDB_CODE_MND_INVALID_STB_OPTION        TAOS_DEF_ERROR_CODE(0, 0x03A5)
+#define TSDB_CODE_MND_INVALID_STB_ALTER_OPTION  TAOS_DEF_ERROR_CODE(0, 0x03A6)
+#define TSDB_CODE_MND_STB_OPTION_UNCHNAGED      TAOS_DEF_ERROR_CODE(0, 0x03A7)
+#define TSDB_CODE_MND_INVALID_ROW_BYTES         TAOS_DEF_ERROR_CODE(0, 0x03A8)
+#define TSDB_CODE_MND_TOO_MANY_TAGS             TAOS_DEF_ERROR_CODE(0, 0x03A9)
+#define TSDB_CODE_MND_TAG_ALREADY_EXIST         TAOS_DEF_ERROR_CODE(0, 0x03AA)
+#define TSDB_CODE_MND_TAG_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x03AB)
+#define TSDB_CODE_MND_TOO_MANY_COLUMNS          TAOS_DEF_ERROR_CODE(0, 0x03AC)
+#define TSDB_CODE_MND_COLUMN_ALREADY_EXIST      TAOS_DEF_ERROR_CODE(0, 0x03AD)
+#define TSDB_CODE_MND_COLUMN_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03AE)
 
 // mnode-func
 #define TSDB_CODE_MND_FUNC_ALREADY_EXIST        TAOS_DEF_ERROR_CODE(0, 0x03C0)

source/common/src/tmsg.c

@@ -27,7 +27,7 @@
 #undef TD_MSG_SEG_CODE_
 #include "tmsgdef.h"
 
-int tInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter) {
+int32_t tInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter) {
   if (pMsg == NULL) {
     terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
     return -1;
@@ -44,7 +44,7 @@ int tInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter) {
   return 0;
 }
 
-int tGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
+int32_t tGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
   if (pIter->len == 0) {
     pIter->len += sizeof(SSubmitMsg);
   } else {
@@ -63,7 +63,7 @@ int tGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
   return 0;
 }
 
-int tInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
+int32_t tInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
   if (pBlock->dataLen <= 0) return -1;
   pIter->totalLen = pBlock->dataLen;
   pIter->len = 0;
@@ -85,14 +85,14 @@ SMemRow tGetSubmitBlkNext(SSubmitBlkIter *pIter) {
   }
 }
 
-int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
-  int tlen = 0;
+int32_t tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
+  int32_t tlen = 0;
   tlen += taosEncodeSClientHbKey(buf, &pReq->connKey);
 
   int32_t kvNum = taosHashGetSize(pReq->info);
   tlen += taosEncodeFixedI32(buf, kvNum);
-  SKv *kv;
-  void* pIter = taosHashIterate(pReq->info, NULL);
+  SKv  *kv;
+  void *pIter = taosHashIterate(pReq->info, NULL);
   while (pIter != NULL) {
     kv = pIter;
     tlen += taosEncodeSKv(buf, kv);
@@ -111,7 +111,7 @@ void *tDeserializeSClientHbReq(void *buf, SClientHbReq *pReq) {
   if (pReq->info == NULL) {
     pReq->info = taosHashInit(kvNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
   }
-  for(int i = 0; i < kvNum; i++) {
+  for (int32_t i = 0; i < kvNum; i++) {
     SKv kv;
     buf = taosDecodeSKv(buf, &kv);
     taosHashPut(pReq->info, &kv.key, sizeof(kv.key), &kv, sizeof(kv));
@@ -120,54 +120,55 @@ void *tDeserializeSClientHbReq(void *buf, SClientHbReq *pReq) {
   return buf;
 }
 
-int   tSerializeSClientHbRsp(void** buf, const SClientHbRsp* pRsp) {
-  int tlen = 0;
+int32_t tSerializeSClientHbRsp(void **buf, const SClientHbRsp *pRsp) {
+  int32_t tlen = 0;
   int32_t kvNum = taosArrayGetSize(pRsp->info);
   tlen += taosEncodeSClientHbKey(buf, &pRsp->connKey);
   tlen += taosEncodeFixedI32(buf, pRsp->status);
   tlen += taosEncodeFixedI32(buf, kvNum);
-  for (int i = 0; i < kvNum; i++) {
+  for (int32_t i = 0; i < kvNum; i++) {
     SKv *kv = (SKv *)taosArrayGet(pRsp->info, i);
     tlen += taosEncodeSKv(buf, kv);
   }
   return tlen;
 }
-void* tDeserializeSClientHbRsp(void* buf, SClientHbRsp* pRsp) {
+
+void *tDeserializeSClientHbRsp(void *buf, SClientHbRsp *pRsp) {
   int32_t kvNum = 0;
   buf = taosDecodeSClientHbKey(buf, &pRsp->connKey);
   buf = taosDecodeFixedI32(buf, &pRsp->status);
   buf = taosDecodeFixedI32(buf, &kvNum);
   pRsp->info = taosArrayInit(kvNum, sizeof(SKv));
-  for (int i = 0; i < kvNum; i++) {
+  for (int32_t i = 0; i < kvNum; i++) {
     SKv kv = {0};
     buf = taosDecodeSKv(buf, &kv);
     taosArrayPush(pRsp->info, &kv);
   }
-  
+
   return buf;
 }
 
-int   tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pBatchReq) {
-  int tlen = 0;
+int32_t tSerializeSClientHbBatchReq(void **buf, const SClientHbBatchReq *pBatchReq) {
+  int32_t tlen = 0;
   tlen += taosEncodeFixedI64(buf, pBatchReq->reqId);
   int32_t reqNum = taosArrayGetSize(pBatchReq->reqs);
-  tlen += taosEncodeFixedI32(buf, reqNum); 
-  for (int i = 0; i < reqNum; i++) {
-    SClientHbReq* pReq = taosArrayGet(pBatchReq->reqs, i);
+  tlen += taosEncodeFixedI32(buf, reqNum);
+  for (int32_t i = 0; i < reqNum; i++) {
+    SClientHbReq *pReq = taosArrayGet(pBatchReq->reqs, i);
     tlen += tSerializeSClientHbReq(buf, pReq);
   }
   return tlen;
 }
 
-void* tDeserializeSClientHbBatchReq(void* buf, SClientHbBatchReq* pBatchReq) {
+void *tDeserializeSClientHbBatchReq(void *buf, SClientHbBatchReq *pBatchReq) {
   buf = taosDecodeFixedI64(buf, &pBatchReq->reqId);
   if (pBatchReq->reqs == NULL) {
     pBatchReq->reqs = taosArrayInit(0, sizeof(SClientHbReq));
   }
- 
+
   int32_t reqNum;
   buf = taosDecodeFixedI32(buf, &reqNum);
-  for (int i = 0; i < reqNum; i++) {
+  for (int32_t i = 0; i < reqNum; i++) {
     SClientHbReq req = {0};
     buf = tDeserializeSClientHbReq(buf, &req);
     taosArrayPush(pBatchReq->reqs, &req);
@@ -175,31 +176,31 @@ void* tDeserializeSClientHbBatchReq(void* buf, SClientHbBatchReq* pBatchReq) {
   return buf;
 }
 
-int   tSerializeSClientHbBatchRsp(void** buf, const SClientHbBatchRsp* pBatchRsp) {
-  int tlen = 0;
+int32_t tSerializeSClientHbBatchRsp(void **buf, const SClientHbBatchRsp *pBatchRsp) {
+  int32_t tlen = 0;
   int32_t sz = taosArrayGetSize(pBatchRsp->rsps);
   tlen += taosEncodeFixedI32(buf, sz);
-  for (int i = 0; i < sz; i++) {
-    SClientHbRsp* pRsp = taosArrayGet(pBatchRsp->rsps, i);
+  for (int32_t i = 0; i < sz; i++) {
+    SClientHbRsp *pRsp = taosArrayGet(pBatchRsp->rsps, i);
     tlen += tSerializeSClientHbRsp(buf, pRsp);
   }
   return tlen;
 }
 
-void* tDeserializeSClientHbBatchRsp(void* buf, SClientHbBatchRsp* pBatchRsp) {
+void *tDeserializeSClientHbBatchRsp(void *buf, SClientHbBatchRsp *pBatchRsp) {
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);
   pBatchRsp->rsps = taosArrayInit(sz, sizeof(SClientHbRsp));
-  for (int i = 0; i < sz; i++) {
+  for (int32_t i = 0; i < sz; i++) {
     SClientHbRsp rsp = {0};
-    buf = tDeserializeSClientHbRsp(buf, &rsp); 
+    buf = tDeserializeSClientHbRsp(buf, &rsp);
     taosArrayPush(pBatchRsp->rsps, &rsp);
   }
   return buf;
 }
 
-int tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
-  int tlen = 0;
+int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
+  int32_t tlen = 0;
 
   tlen += taosEncodeFixedU64(buf, pReq->ver);
   tlen += taosEncodeString(buf, pReq->name);
@@ -293,8 +294,8 @@ void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
   return buf;
 }
 
-int tSerializeSVCreateTbBatchReq(void **buf, SVCreateTbBatchReq *pReq) {
-  int tlen = 0;
+int32_t tSerializeSVCreateTbBatchReq(void **buf, SVCreateTbBatchReq *pReq) {
+  int32_t tlen = 0;
 
   tlen += taosEncodeFixedU64(buf, pReq->ver);
   tlen += taosEncodeFixedU32(buf, taosArrayGetSize(pReq->pArray));
@@ -322,7 +323,7 @@ void *tDeserializeSVCreateTbBatchReq(void *buf, SVCreateTbBatchReq *pReq) {
 }
 
 int32_t tSerializeSVDropTbReq(void **buf, SVDropTbReq *pReq) {
-  int tlen = 0;
+  int32_t tlen = 0;
   tlen += taosEncodeFixedU64(buf, pReq->ver);
   tlen += taosEncodeString(buf, pReq->name);
   tlen += taosEncodeFixedU8(buf, pReq->type);

source/dnode/mgmt/impl/test/vnode/vnode.cpp

@@ -108,7 +108,7 @@ TEST_F(DndTestVnode, 01_Create_Vnode) {
   }
 }
 
-TEST_F(DndTestVnode, 02_ALTER_Vnode) {
+TEST_F(DndTestVnode, 02_Alter_Vnode) {
   for (int i = 0; i < 3; ++i) {
     int32_t contLen = sizeof(SAlterVnodeReq);
 
@@ -219,7 +219,7 @@ TEST_F(DndTestVnode, 03_Create_Stb) {
   }
 }
 
-TEST_F(DndTestVnode, 04_ALTER_Stb) {
+TEST_F(DndTestVnode, 04_Alter_Stb) {
   for (int i = 0; i < 1; ++i) {
     SVCreateTbReq req = {0};
     req.ver = 0;
@@ -310,7 +310,7 @@ TEST_F(DndTestVnode, 05_DROP_Stb) {
   }
 }
 
-TEST_F(DndTestVnode, 06_DROP_Vnode) {
+TEST_F(DndTestVnode, 06_Drop_Vnode) {
   for (int i = 0; i < 3; ++i) {
     int32_t contLen = sizeof(SDropVnodeReq);
 

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

@@ -652,12 +652,12 @@ static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, const SSc
 
   for (int32_t i = 0; i < ntags; i++) {
     if (mndFindSuperTableColumnIndex(pOld, pSchemas[i].name) > 0) {
-      terrno = TSDB_CODE_MND_TAG_ALREAY_EXIST;
+      terrno = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
       return -1;
     }
 
     if (mndFindSuperTableTagIndex(pOld, pSchemas[i].name) > 0) {
-      terrno = TSDB_CODE_MND_COLUMN_ALREAY_EXIST;
+      terrno = TSDB_CODE_MND_TAG_ALREADY_EXIST;
       return -1;
     }
   }
@@ -692,6 +692,7 @@ static int32_t mndDropSuperTableTag(const SStbObj *pOld, SStbObj *pNew, const ch
   }
 
   memmove(pNew->pTags + tag, pNew->pTags + tag + 1, sizeof(SSchema) * (pNew->numOfTags - tag - 1));
+  pNew->numOfTags--;
 
   pNew->version++;
   mDebug("stb:%s, start to drop tag %s", pNew->name, tagName);
@@ -706,13 +707,12 @@ static int32_t mndAlterStbTagName(const SStbObj *pOld, SStbObj *pNew, const char
   }
 
   if (mndFindSuperTableTagIndex(pOld, newTagName) >= 0) {
-    terrno = TSDB_CODE_MND_TAG_ALREAY_EXIST;
+    terrno = TSDB_CODE_MND_TAG_ALREADY_EXIST;
     return -1;
   }
 
-  int32_t len = (int32_t)strlen(newTagName);
-  if (len >= TSDB_COL_NAME_LEN) {
-    terrno = TSDB_CODE_MND_INVALID_STB_OPTION;
+  if (mndFindSuperTableColumnIndex(pOld, newTagName) >= 0) {
+    terrno = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
     return -1;
   }
 
@@ -735,16 +735,17 @@ static int32_t mndAlterStbTagBytes(const SStbObj *pOld, SStbObj *pNew, const SSc
     return -1;
   }
 
-  if (!(pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR)) {
-    terrno = TSDB_CODE_MND_INVALID_STB_OPTION;
+  if (mndAllocStbSchemas(pOld, pNew) != 0) {
     return -1;
   }
 
-  if (mndAllocStbSchemas(pOld, pNew) != 0) {
+  SSchema *pTag = pNew->pTags + tag;
+
+  if (!(pTag->type == TSDB_DATA_TYPE_BINARY || pTag->type == TSDB_DATA_TYPE_NCHAR)) {
+    terrno = TSDB_CODE_MND_INVALID_STB_OPTION;
     return -1;
   }
 
-  SSchema *pTag = pNew->pTags + tag;
   if (pSchema->bytes <= pTag->bytes) {
     terrno = TSDB_CODE_MND_INVALID_ROW_BYTES;
     return -1;
@@ -765,12 +766,12 @@ static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, const
 
   for (int32_t i = 0; i < ncols; i++) {
     if (mndFindSuperTableColumnIndex(pOld, pSchemas[i].name) > 0) {
-      terrno = TSDB_CODE_MND_TAG_ALREAY_EXIST;
+      terrno = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
       return -1;
     }
 
     if (mndFindSuperTableTagIndex(pOld, pSchemas[i].name) > 0) {
-      terrno = TSDB_CODE_MND_COLUMN_ALREAY_EXIST;
+      terrno = TSDB_CODE_MND_TAG_ALREADY_EXIST;
       return -1;
     }
   }
@@ -795,16 +796,27 @@ static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, const
 
 static int32_t mndDropSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, const char *colName) {
   int32_t col = mndFindSuperTableColumnIndex(pOld, colName);
-  if (col <= 0) {
+  if (col < 0) {
     terrno = TSDB_CODE_MND_COLUMN_NOT_EXIST;
     return -1;
   }
 
+  if (col == 0) {
+    terrno = TSDB_CODE_MND_INVALID_STB_ALTER_OPTION;
+    return -1;
+  }
+
+  if (pOld->numOfColumns == 2) {
+    terrno = TSDB_CODE_MND_INVALID_STB_ALTER_OPTION;
+    return -1;
+  }
+
   if (mndAllocStbSchemas(pOld, pNew) != 0) {
     return -1;
   }
 
   memmove(pNew->pColumns + col, pNew->pColumns + col + 1, sizeof(SSchema) * (pNew->numOfColumns - col - 1));
+  pNew->numOfColumns--;
 
   pNew->version++;
   mDebug("stb:%s, start to drop col %s", pNew->name, colName);
@@ -818,11 +830,6 @@ static int32_t mndAlterStbColumnBytes(const SStbObj *pOld, SStbObj *pNew, const
     return -1;
   }
 
-  if (!(pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR)) {
-    terrno = TSDB_CODE_MND_INVALID_STB_OPTION;
-    return -1;
-  }
-
   uint32_t nLen = 0;
   for (int32_t i = 0; i < pOld->numOfColumns; ++i) {
     nLen += (pOld->pColumns[i].colId == col) ? pSchema->bytes : pOld->pColumns[i].bytes;
@@ -838,6 +845,11 @@ static int32_t mndAlterStbColumnBytes(const SStbObj *pOld, SStbObj *pNew, const
   }
 
   SSchema *pCol = pNew->pColumns + col;
+  if (!(pCol->type == TSDB_DATA_TYPE_BINARY || pCol->type == TSDB_DATA_TYPE_NCHAR)) {
+    terrno = TSDB_CODE_MND_INVALID_STB_OPTION;
+    return -1;
+  }
+
   if (pSchema->bytes <= pCol->bytes) {
     terrno = TSDB_CODE_MND_INVALID_ROW_BYTES;
     return -1;
@@ -914,12 +926,13 @@ static int32_t mndAlterStb(SMnode *pMnode, SMnodeMsg *pReq, const SMAltertbReq *
   taosRUnLockLatch(&pOld->lock);
 
   int32_t code = -1;
+  STrans *pTrans = NULL;
 
   switch (pAlter->alterType) {
-    case TSDB_ALTER_TABLE_ADD_TAG_COLUMN:
+    case TSDB_ALTER_TABLE_ADD_TAG:
       code = mndAddSuperTableTag(pOld, &stbObj, pAlter->pSchemas, 1);
       break;
-    case TSDB_ALTER_TABLE_DROP_TAG_COLUMN:
+    case TSDB_ALTER_TABLE_DROP_TAG:
       code = mndDropSuperTableTag(pOld, &stbObj, pAlter->pSchemas[0].name);
       break;
     case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:
@@ -945,7 +958,7 @@ static int32_t mndAlterStb(SMnode *pMnode, SMnodeMsg *pReq, const SMAltertbReq *
   if (code != 0) goto ALTER_STB_OVER;
 
   code = -1;
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pReq->rpcMsg);
+  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pReq->rpcMsg);
   if (pTrans == NULL) goto ALTER_STB_OVER;
 
   mDebug("trans:%d, used to alter stb:%s", pTrans->id, pAlter->name);
@@ -975,17 +988,17 @@ static int32_t mndProcessMAlterStbReq(SMnodeMsg *pReq) {
     return -1;
   }
 
-  SStbObj *pStb = mndAcquireStb(pMnode, pAlter->name);
-  if (pStb == NULL) {
-    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
+  SDbObj *pDb = mndAcquireDbByStb(pMnode, pAlter->name);
+  if (pDb == NULL) {
+    terrno = TSDB_CODE_MND_INVALID_DB;
     mError("stb:%s, failed to alter since %s", pAlter->name, terrstr());
     return -1;
   }
 
-  SDbObj *pDb = mndAcquireDbByStb(pMnode, pAlter->name);
-  if (pDb == NULL) {
-    mndReleaseStb(pMnode, pStb);
-    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
+  SStbObj *pStb = mndAcquireStb(pMnode, pAlter->name);
+  if (pStb == NULL) {
+    mndReleaseDb(pMnode, pDb);
+    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
     mError("stb:%s, failed to alter since %s", pAlter->name, terrstr());
     return -1;
   }
@@ -994,7 +1007,7 @@ static int32_t mndProcessMAlterStbReq(SMnodeMsg *pReq) {
   mndReleaseStb(pMnode, pStb);
 
   if (code != 0) {
-    mError("stb:%s, failed to alter since %s", pAlter->name, tstrerror(code));
+    mError("stb:%s, failed to alter since %s", pAlter->name, terrstr());
     return code;
   }
 

source/libs/parser/inc/sql.y

@@ -820,7 +820,7 @@ cmd ::= ALTER TABLE ids(X) cpxName(F) MODIFY COLUMN columnlist(A).     {
 //////////////////////////////////ALTER TAGS statement/////////////////////////////////////
 cmd ::= ALTER TABLE ids(X) cpxName(Y) ADD TAG columnlist(A).        {
     X.n += Y.n;
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, A, NULL, TSDB_ALTER_TABLE_ADD_TAG_COLUMN, -1);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, A, NULL, TSDB_ALTER_TABLE_ADD_TAG, -1);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
 cmd ::= ALTER TABLE ids(X) cpxName(Z) DROP TAG ids(Y).          {
@@ -829,7 +829,7 @@ cmd ::= ALTER TABLE ids(X) cpxName(Z) DROP TAG ids(Y).          {
     toTSDBType(Y.type);
     SArray* A = tListItemAppendToken(NULL, &Y, -1);
 
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, NULL, A, TSDB_ALTER_TABLE_DROP_TAG_COLUMN, -1);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, NULL, A, TSDB_ALTER_TABLE_DROP_TAG, -1);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
 
@@ -889,7 +889,7 @@ cmd ::= ALTER STABLE ids(X) cpxName(F) MODIFY COLUMN columnlist(A).     {
 //////////////////////////////////ALTER TAGS statement/////////////////////////////////////
 cmd ::= ALTER STABLE ids(X) cpxName(Y) ADD TAG columnlist(A).        {
     X.n += Y.n;
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, A, NULL, TSDB_ALTER_TABLE_ADD_TAG_COLUMN, TSDB_SUPER_TABLE);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, A, NULL, TSDB_ALTER_TABLE_ADD_TAG, TSDB_SUPER_TABLE);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
 cmd ::= ALTER STABLE ids(X) cpxName(Z) DROP TAG ids(Y).          {
@@ -898,7 +898,7 @@ cmd ::= ALTER STABLE ids(X) cpxName(Z) DROP TAG ids(Y).          {
     toTSDBType(Y.type);
     SArray* A = tListItemAppendToken(NULL, &Y, -1);
 
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, NULL, A, TSDB_ALTER_TABLE_DROP_TAG_COLUMN, TSDB_SUPER_TABLE);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&X, NULL, A, TSDB_ALTER_TABLE_DROP_TAG, TSDB_SUPER_TABLE);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
 

source/libs/parser/src/astGenerator.c

@@ -610,7 +610,7 @@ SAlterTableInfo *tSetAlterTableInfo(SToken *pTableName, SArray *pCols, SArray *p
   pAlterTable->type = type;
   pAlterTable->tableType = tableType;
 
-  if (type == TSDB_ALTER_TABLE_ADD_COLUMN || type == TSDB_ALTER_TABLE_ADD_TAG_COLUMN || type == TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES || type == TSDB_ALTER_TABLE_UPDATE_TAG_BYTES) {
+  if (type == TSDB_ALTER_TABLE_ADD_COLUMN || type == TSDB_ALTER_TABLE_ADD_TAG || type == TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES || type == TSDB_ALTER_TABLE_UPDATE_TAG_BYTES) {
     pAlterTable->pAddColumns = pCols;
     assert(pVals == NULL);
   } else {

source/libs/parser/src/sql.c

@@ -3206,7 +3206,7 @@ static void yy_reduce(
       case 286: /* cmd ::= ALTER TABLE ids cpxName ADD TAG columnlist */
 {
     yymsp[-4].minor.yy0.n += yymsp[-3].minor.yy0.n;
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, yymsp[0].minor.yy165, NULL, TSDB_ALTER_TABLE_ADD_TAG_COLUMN, -1);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, yymsp[0].minor.yy165, NULL, TSDB_ALTER_TABLE_ADD_TAG, -1);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
         break;
@@ -3217,7 +3217,7 @@ static void yy_reduce(
     toTSDBType(yymsp[0].minor.yy0.type);
     SArray* A = tListItemAppendToken(NULL, &yymsp[0].minor.yy0, -1);
 
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, NULL, A, TSDB_ALTER_TABLE_DROP_TAG_COLUMN, -1);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, NULL, A, TSDB_ALTER_TABLE_DROP_TAG, -1);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
         break;
@@ -3282,7 +3282,7 @@ static void yy_reduce(
       case 294: /* cmd ::= ALTER STABLE ids cpxName ADD TAG columnlist */
 {
     yymsp[-4].minor.yy0.n += yymsp[-3].minor.yy0.n;
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, yymsp[0].minor.yy165, NULL, TSDB_ALTER_TABLE_ADD_TAG_COLUMN, TSDB_SUPER_TABLE);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, yymsp[0].minor.yy165, NULL, TSDB_ALTER_TABLE_ADD_TAG, TSDB_SUPER_TABLE);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
         break;
@@ -3293,7 +3293,7 @@ static void yy_reduce(
     toTSDBType(yymsp[0].minor.yy0.type);
     SArray* A = tListItemAppendToken(NULL, &yymsp[0].minor.yy0, -1);
 
-    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, NULL, A, TSDB_ALTER_TABLE_DROP_TAG_COLUMN, TSDB_SUPER_TABLE);
+    SAlterTableInfo* pAlterTable = tSetAlterTableInfo(&yymsp[-4].minor.yy0, NULL, A, TSDB_ALTER_TABLE_DROP_TAG, TSDB_SUPER_TABLE);
     setSqlInfo(pInfo, pAlterTable, NULL, TSDB_SQL_ALTER_TABLE);
 }
         break;

source/libs/sync/CMakeLists.txt

+
+aux_source_directory(src SYNC_SRC)
+add_library(sync ${SYNC_SRC})
+
+target_link_libraries(
+    sync
+    PUBLIC common
+    PUBLIC transport
+    PUBLIC util
+    PUBLIC wal
+)
+
+target_include_directories(
+    sync
+    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
\ No newline at end of file

source/libs/sync/src/sync.c

+#include "sync.h"
\ No newline at end of file

source/util/src/terror.c

@@ -228,17 +228,19 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_VGROUP_NOT_EXIST,         "VGroup does not exist
 // mnode-stable
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_STB_ALREADY_EXIST,        "Stable already exists")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_STB_NOT_EXIST,            "Stable not exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_NAME_CONFLICT_WITH_TOPIC, "Stable confilct with topic")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_STBS,            "Too many stables")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STB,              "Invalid stable name")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STB_OPTION,       "Invalid stable options")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_STB_OPTION_UNCHNAGED,     "Stable options not changed")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STB_ALTER_OPTION, "Invalid stable alter options")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_STB_OPTION_UNCHNAGED,     "Stable option unchanged")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ROW_BYTES,        "Invalid row bytes")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TAGS,            "Too many tags")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_TAG_ALREAY_EXIST,         "Tag already exists")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_TAG_ALREADY_EXIST,        "Tag already exists")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TAG_NOT_EXIST,            "Tag does not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_COLUMNS,         "Too many columns")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_COLUMN_ALREAY_EXIST,      "Column already exists")
+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_INVALID_ROW_BYTES,        "Invalid row bytes")
 
 // mnode-func
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_FUNC_ALREADY_EXIST,       "Func already exists")

grafanaplugin @ 4a4d7909

-Subproject commit 4a4d79099b076b8ff12d5b4fdbcba54049a6866d