Merge branch '3.0' of github.com:taosdata/TDengine into 3.0

include/common/tmsg.h

@@ -147,6 +147,11 @@ typedef enum _mgmt_table {
 #define TSDB_COL_IS_NORMAL_COL(f) ((f & (~(TSDB_COL_NULL))) == TSDB_COL_NORMAL)
 #define TSDB_COL_IS_UD_COL(f) ((f & (~(TSDB_COL_NULL))) == TSDB_COL_UDC)
 #define TSDB_COL_REQ_NULL(f) (((f)&TSDB_COL_NULL) != 0)
+
+#define TD_SUPER_TABLE TSDB_SUPER_TABLE
+#define TD_CHILD_TABLE TSDB_CHILD_TABLE
+#define TD_NORMAL_TABLE TSDB_NORMAL_TABLE
+
 typedef struct {
   int32_t vgId;
   char*   dbName;
@@ -1154,10 +1159,7 @@ typedef struct SVCreateTbReq {
   char*    name;
   uint32_t ttl;
   uint32_t keep;
-#define TD_SUPER_TABLE TSDB_SUPER_TABLE
-#define TD_CHILD_TABLE TSDB_CHILD_TABLE
-#define TD_NORMAL_TABLE TSDB_NORMAL_TABLE
-  uint8_t type;
+  uint8_t  type;
   union {
     struct {
       tb_uid_t suid;
@@ -1202,15 +1204,21 @@ typedef struct {
 } SVAlterTbRsp;
 
 typedef struct {
-  SMsgHead head;
-  char     name[TSDB_TABLE_FNAME_LEN];
-  int64_t  suid;
+  uint64_t ver;
+  char*    name;
+  uint8_t  type;
+  tb_uid_t suid;
 } SVDropTbReq;
 
 typedef struct {
-  SMsgHead head;
+  uint64_t ver;
 } SVDropTbRsp;
 
+int32_t tSerializeSVDropTbReq(void** buf, SVDropTbReq* pReq);
+void*   tDeserializeSVDropTbReq(void* buf, SVDropTbReq* pReq);
+int32_t tSerializeSVDropTbRsp(void** buf, SVDropTbRsp* pRsp);
+void*   tDeserializeSVDropTbRsp(void* buf, SVDropTbRsp* pRsp);
+
 typedef struct {
   SMsgHead head;
   int64_t  uid;

source/common/src/tmsg.c

@@ -320,3 +320,18 @@ void *tSVCreateTbBatchReqDeserialize(void *buf, SVCreateTbBatchReq *pReq) {
 
   return buf;
 }
+
+int32_t tSerializeSVDropTbReq(void **buf, SVDropTbReq *pReq) {
+  int tlen = 0;
+  tlen += taosEncodeFixedU64(buf, pReq->ver);
+  tlen += taosEncodeString(buf, pReq->name);
+  tlen += taosEncodeFixedU8(buf, pReq->type);
+  return tlen;
+}
+
+void *tDeserializeSVDropTbReq(void *buf, SVDropTbReq *pReq) {
+  buf = taosDecodeFixedU64(buf, &pReq->ver);
+  buf = taosDecodeString(buf, &pReq->name);
+  buf = taosDecodeFixedU8(buf, &pReq->type);
+  return buf;
+}

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

@@ -199,19 +199,16 @@ TEST_F(DndTestVnode, 03_Create_Stb) {
     req.stbCfg.nTagCols = 3;
     req.stbCfg.pTagSchema = &schemas[2];
 
-    int32_t   bsize = tSerializeSVCreateTbReq(NULL, &req);
-    void*     buf = rpcMallocCont(sizeof(SMsgHead) + bsize);
-    SMsgHead* pMsgHead = (SMsgHead*)buf;
+    int32_t   contLen = tSerializeSVCreateTbReq(NULL, &req) + sizeof(SMsgHead);
+    SMsgHead* pHead = (SMsgHead*)rpcMallocCont(contLen);
 
-    pMsgHead->contLen = htonl(sizeof(SMsgHead) + bsize);
-    pMsgHead->vgId = htonl(2);
+    pHead->contLen = htonl(contLen);
+    pHead->vgId = htonl(2);
 
-    void* pBuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
+    void* pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
     tSerializeSVCreateTbReq(&pBuf, &req);
 
-    int32_t contLen = sizeof(SMsgHead) + bsize;
-
-    SRpcMsg* pRsp = test.SendReq(TDMT_VND_CREATE_STB, buf, contLen);
+    SRpcMsg* pRsp = test.SendReq(TDMT_VND_CREATE_STB, (void*)pHead, contLen);
     ASSERT_NE(pRsp, nullptr);
     if (i == 0) {
       ASSERT_EQ(pRsp->code, 0);
@@ -235,20 +232,28 @@ TEST_F(DndTestVnode, 04_ALTER_Stb) {
 }
 
 TEST_F(DndTestVnode, 05_DROP_Stb) {
-#if 0
   {
     for (int i = 0; i < 3; ++i) {
-      SRpcMsg* pRsp = test.SendReq(TDMT_VND_DROP_STB, pReq, contLen);
+      SVDropTbReq req = {0};
+      req.ver = 0;
+      req.name = (char*)"stb1";
+      req.suid = 9599;
+      req.type = TD_SUPER_TABLE;
+
+      int32_t   contLen = tSerializeSVDropTbReq(NULL, &req) + sizeof(SMsgHead);
+      SMsgHead* pHead = (SMsgHead*)rpcMallocCont(contLen);
+
+      pHead->contLen = htonl(contLen);
+      pHead->vgId = htonl(2);
+
+      void* pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
+      tSerializeSVDropTbReq(&pBuf, &req);
+
+      SRpcMsg* pRsp = test.SendReq(TDMT_VND_DROP_STB, (void*)pHead, contLen);
       ASSERT_NE(pRsp, nullptr);
-      if (i == 0) {
-        ASSERT_EQ(pRsp->code, 0);
-        test.Restart();
-      } else {
-        ASSERT_EQ(pRsp->code, TSDB_CODE_TDB_INVALID_TABLE_ID);
-      }
+      ASSERT_EQ(pRsp->code, 0);
     }
   }
-#endif
 }
 
 TEST_F(DndTestVnode, 06_DROP_Vnode) {

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

@@ -181,23 +181,26 @@ static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {
 
 static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
   mTrace("stb:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
-  atomic_exchange_32(&pOld->updateTime, pNew->updateTime);
-  atomic_exchange_32(&pOld->version, pNew->version);
 
   taosWLockLatch(&pOld->lock);
-  pOld->numOfColumns = pNew->numOfColumns;
-  pOld->numOfTags = pNew->numOfTags;
   int32_t totalCols = pNew->numOfTags + pNew->numOfColumns;
   int32_t totalSize = totalCols * sizeof(SSchema);
-
   if (pOld->numOfTags + pOld->numOfColumns < totalCols) {
     void *pSchema = malloc(totalSize);
     if (pSchema != NULL) {
       free(pOld->pSchema);
       pOld->pSchema = pSchema;
+    } else {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
+      taosWUnLockLatch(&pOld->lock);
     }
   }
 
+  pOld->updateTime = pNew->updateTime;
+  pOld->version = pNew->version;
+  pOld->numOfColumns = pNew->numOfColumns;
+  pOld->numOfTags = pNew->numOfTags;
   memcpy(pOld->pSchema, pNew->pSchema, totalSize);
   taosWUnLockLatch(&pOld->lock);
   return 0;
@@ -228,15 +231,11 @@ static SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {
 }
 
 static void *mndBuildCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
-  SVCreateTbReq req;
-  void         *buf;
-  int32_t       bsize;
-  SMsgHead     *pMsgHead;
-
-  req.ver = 0;
   SName name = {0};
   tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
 
+  SVCreateTbReq req = {0};
+  req.ver = 0;
   req.name = (char *)tNameGetTableName(&name);
   req.ttl = 0;
   req.keep = 0;
@@ -247,40 +246,48 @@ static void *mndBuildCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb
   req.stbCfg.nTagCols = pStb->numOfTags;
   req.stbCfg.pTagSchema = pStb->pSchema + pStb->numOfColumns;
 
-  bsize = tSerializeSVCreateTbReq(NULL, &req);
-  buf = malloc(sizeof(SMsgHead) + bsize);
-  if (buf == NULL) {
+  int32_t   contLen = tSerializeSVCreateTbReq(NULL, &req) + sizeof(SMsgHead);
+  SMsgHead *pHead = malloc(contLen);
+  if (pHead == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
 
-  pMsgHead = (SMsgHead *)buf;
-
-  pMsgHead->contLen = htonl(sizeof(SMsgHead) + bsize);
-  pMsgHead->vgId = htonl(pVgroup->vgId);
+  pHead->contLen = htonl(contLen);
+  pHead->vgId = htonl(pVgroup->vgId);
 
-  void *pBuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
+  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
   tSerializeSVCreateTbReq(&pBuf, &req);
 
-  *pContLen = sizeof(SMsgHead) + bsize;
-  return buf;
+  *pContLen = contLen;
+  return pHead;
 }
 
-static SVDropTbReq *mndBuildDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb) {
-  int32_t contLen = sizeof(SVDropTbReq);
+static void *mndBuildDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
+  SName name = {0};
+  tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
+
+  SVDropTbReq req = {0};
+  req.ver = 0;
+  req.name = (char *)tNameGetTableName(&name);
+  req.type = TD_SUPER_TABLE;
+  req.suid = pStb->uid;
 
-  SVDropTbReq *pDrop = calloc(1, contLen);
-  if (pDrop == NULL) {
+  int32_t   contLen = tSerializeSVDropTbReq(NULL, &req) + sizeof(SMsgHead);
+  SMsgHead *pHead = malloc(contLen);
+  if (pHead == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
 
-  pDrop->head.contLen = htonl(contLen);
-  pDrop->head.vgId = htonl(pVgroup->vgId);
-  memcpy(pDrop->name, pStb->name, TSDB_TABLE_FNAME_LEN);
-  pDrop->suid = htobe64(pStb->uid);
+  pHead->contLen = htonl(contLen);
+  pHead->vgId = htonl(pVgroup->vgId);
 
-  return pDrop;
+  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
+  tSerializeSVDropTbReq(&pBuf, &req);
+
+  *pContLen = contLen;
+  return pHead;
 }
 
 static int32_t mndCheckCreateStbReq(SMCreateStbReq *pCreate) {
@@ -358,7 +365,7 @@ static int32_t mndSetCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
   SSdb   *pSdb = pMnode->pSdb;
   SVgObj *pVgroup = NULL;
   void   *pIter = NULL;
-  int32_t     contLen;
+  int32_t contLen;
 
   while (1) {
     pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
@@ -400,7 +407,8 @@ static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
     if (pIter == NULL) break;
     if (pVgroup->dbUid != pDb->uid) continue;
 
-    SVDropTbReq *pReq = mndBuildDropStbReq(pMnode, pVgroup, pStb);
+    int32_t contLen = 0;
+    void   *pReq = mndBuildDropStbReq(pMnode, pVgroup, pStb, &contLen);
     if (pReq == NULL) {
       sdbCancelFetch(pSdb, pIter);
       sdbRelease(pSdb, pVgroup);
@@ -411,7 +419,7 @@ static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
     STransAction action = {0};
     action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
     action.pCont = pReq;
-    action.contLen = sizeof(SVDropTbReq);
+    action.contLen = contLen;
     action.msgType = TDMT_VND_DROP_STB;
     if (mndTransAppendUndoAction(pTrans, &action) != 0) {
       free(pReq);
@@ -498,9 +506,9 @@ static int32_t mndProcessMCreateStbReq(SMnodeMsg *pReq) {
   }
 
   // topic should have different name with stb
-  SStbObj *pTopic = mndAcquireStb(pMnode, pCreate->name);
-  if (pTopic != NULL) {
-    sdbRelease(pMnode->pSdb, pTopic);
+  SStbObj *pTopicStb = mndAcquireStb(pMnode, pCreate->name);
+  if (pTopicStb != NULL) {
+    mndReleaseStb(pMnode, pTopicStb);
     terrno = TSDB_CODE_MND_NAME_CONFLICT_WITH_TOPIC;
     mError("stb:%s, failed to create since %s", pCreate->name, terrstr());
     return -1;
@@ -517,7 +525,6 @@ static int32_t mndProcessMCreateStbReq(SMnodeMsg *pReq) {
   mndReleaseDb(pMnode, pDb);
 
   if (code != 0) {
-    terrno = code;
     mError("stb:%s, failed to create since %s", pCreate->name, terrstr());
     return -1;
   }
@@ -603,15 +610,6 @@ static int32_t mndSetDropStbRedoLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pS
   return 0;
 }
 
-static int32_t mndSetDropStbUndoLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
-  SSdbRaw *pUndoRaw = mndStbActionEncode(pStb);
-  if (pUndoRaw == NULL) return -1;
-  if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) return -1;
-  if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY) != 0) return -1;
-
-  return 0;
-}
-
 static int32_t mndSetDropStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
   SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
   if (pCommitRaw == NULL) return -1;
@@ -621,22 +619,54 @@ static int32_t mndSetDropStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *
   return 0;
 }
 
-static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) { return 0; }
+static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
+  SSdb   *pSdb = pMnode->pSdb;
+  SVgObj *pVgroup = NULL;
+  void   *pIter = NULL;
+  int32_t contLen;
 
-static int32_t mndSetDropStbUndoActions(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) { return 0; }
+  while (1) {
+    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
+    if (pIter == NULL) break;
+    if (pVgroup->dbUid != pDb->uid) continue;
 
-static int32_t mndDropStb(SMnode *pMnode, SMnodeMsg *pReq, SStbObj *pStb) {
+    int32_t contLen = 0;
+    void   *pReq = mndBuildDropStbReq(pMnode, pVgroup, pStb, &contLen);
+    if (pReq == NULL) {
+      sdbCancelFetch(pSdb, pIter);
+      sdbRelease(pSdb, pVgroup);
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return -1;
+    }
+
+    STransAction action = {0};
+    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
+    action.pCont = pReq;
+    action.contLen = contLen;
+    action.msgType = TDMT_VND_DROP_STB;
+    action.acceptableCode = TSDB_CODE_VND_TB_NOT_EXIST;
+    if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+      free(pReq);
+      sdbCancelFetch(pSdb, pIter);
+      sdbRelease(pSdb, pVgroup);
+      return -1;
+    }
+    sdbRelease(pSdb, pVgroup);
+  }
+
+  return 0;
+}
+
+static int32_t mndDropStb(SMnode *pMnode, SMnodeMsg *pReq, SDbObj *pDb, SStbObj *pStb) {
   int32_t code = -1;
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pReq->rpcMsg);
-  if (pTrans == NULL)goto DROP_STB_OVER;
+  if (pTrans == NULL) goto DROP_STB_OVER;
 
   mDebug("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);
 
   if (mndSetDropStbRedoLogs(pMnode, pTrans, pStb) != 0) goto DROP_STB_OVER;
-  if (mndSetDropStbUndoLogs(pMnode, pTrans, pStb) != 0) goto DROP_STB_OVER;
   if (mndSetDropStbCommitLogs(pMnode, pTrans, pStb) != 0) goto DROP_STB_OVER;
-  if (mndSetDropStbRedoActions(pMnode, pTrans, pStb) != 0) goto DROP_STB_OVER;
-  if (mndSetDropStbUndoActions(pMnode, pTrans, pStb) != 0) goto DROP_STB_OVER;
+  if (mndSetDropStbRedoActions(pMnode, pTrans, pDb, pStb) != 0) goto DROP_STB_OVER;
   if (mndTransPrepare(pMnode, pTrans) != 0) goto DROP_STB_OVER;
 
   code = 0;
@@ -664,7 +694,16 @@ static int32_t mndProcessMDropStbReq(SMnodeMsg *pReq) {
     }
   }
 
-  int32_t code = mndDropStb(pMnode, pReq, pStb);
+  SDbObj *pDb = mndAcquireDbByStb(pMnode, pDrop->name);
+  if (pDb == NULL) {
+    mndReleaseStb(pMnode, pStb);
+    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
+    mError("stb:%s, failed to drop since %s", pDrop->name, terrstr());
+    return -1;
+  }
+
+  int32_t code = mndDropStb(pMnode, pReq, pDb, pStb);
+  mndReleaseDb(pMnode, pDb);
   mndReleaseStb(pMnode, pStb);
 
   if (code != 0) {
@@ -856,7 +895,7 @@ static int32_t mndRetrieveStb(SMnodeMsg *pReq, SShowObj *pShow, char *data, int3
 
     if (pStb->dbUid != pDb->uid) {
       if (strncmp(pStb->db, pDb->name, tListLen(pStb->db)) == 0) {
-        mError("Inconsistent table data, name:%s, db:%s, dbUid:%"PRIu64, pStb->name, pDb->name, pDb->uid);
+        mError("Inconsistent table data, name:%s, db:%s, dbUid:%" PRIu64, pStb->name, pDb->name, pDb->uid);
       }
 
       sdbRelease(pSdb, pStb);

source/dnode/vnode/src/vnd/vnodeWrite.c

@@ -103,7 +103,12 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       taosArrayDestroy(vCreateTbBatchReq.pArray);
       break;
 
+    case TDMT_VND_ALTER_STB:
+      vTrace("vgId:%d, process drop stb req", pVnode->vgId);
+      break;
     case TDMT_VND_DROP_STB:
+      vTrace("vgId:%d, process drop stb req", pVnode->vgId);
+      break;
     case TDMT_VND_DROP_TABLE:
       // if (metaDropTable(pVnode->pMeta, vReq.dtReq.uid) < 0) {
       //   // TODO: handle error

source/util/test/CMakeLists.txt

@@ -45,4 +45,10 @@ target_link_libraries(freelistTest os util gtest gtest_main)
 add_executable(encodeTest "encodeTest.cpp")
 target_link_libraries(encodeTest os util gtest gtest_main)
 
-
+# queueTest
+add_executable(queue_test "queueTest.cpp")
+target_link_libraries(queue_test os util gtest_main)   
+add_test(
+    NAME queue_test
+    COMMAND queue_test
+)
\ No newline at end of file

source/util/test/queueTest.cpp

+/**
+ * @file queue.cpp
+ * @author slguan (slguan@taosdata.com)
+ * @brief UTIL module queue tests
+ * @version 1.0
+ * @date 2022-01-27
+ *
+ * @copyright Copyright (c) 2022
+ *
+ */
+
+#include <gtest/gtest.h>
+
+#include "os.h"
+#include "tqueue.h"
+
+class UtilTestQueue : public ::testing::Test {
+ public:
+  void SetUp() override {}
+  void TearDown() override {}
+
+ public:
+  static void SetUpTestSuite() {}
+  static void TearDownTestSuite() {}
+};
+
+TEST_F(UtilTestQueue, 01_ReadQitemFromQsetByThread) {
+  EXPECT_EQ(0, 0);
+}
\ No newline at end of file