diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 154973e20f..db184b18cd 100644 --- a/include/common/tmsg.h +++ b/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; diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 34b55fd812..3fabd2ce0d 100644 --- a/source/common/src/tmsg.c +++ b/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; +} diff --git a/source/dnode/mgmt/impl/test/vnode/vnode.cpp b/source/dnode/mgmt/impl/test/vnode/vnode.cpp index 17092ed737..11b32fbf0f 100644 --- a/source/dnode/mgmt/impl/test/vnode/vnode.cpp +++ b/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) { diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index 45b8428663..111f092dc3 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/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; + pHead->contLen = htonl(contLen); + pHead->vgId = htonl(pVgroup->vgId); - pMsgHead->contLen = htonl(sizeof(SMsgHead) + bsize); - pMsgHead->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 *pDrop = calloc(1, contLen); - if (pDrop == NULL) { + SVDropTbReq req = {0}; + req.ver = 0; + req.name = (char *)tNameGetTableName(&name); + req.type = TD_SUPER_TABLE; + req.suid = pStb->uid; + + 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); diff --git a/source/dnode/vnode/src/vnd/vnodeWrite.c b/source/dnode/vnode/src/vnd/vnodeWrite.c index 326d99ddbb..1f5513f581 100644 --- a/source/dnode/vnode/src/vnd/vnodeWrite.c +++ b/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 diff --git a/source/util/test/CMakeLists.txt b/source/util/test/CMakeLists.txt index 383a00232a..ee0ade03b8 100644 --- a/source/util/test/CMakeLists.txt +++ b/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 diff --git a/source/util/test/queueTest.cpp b/source/util/test/queueTest.cpp new file mode 100644 index 0000000000..310ae4350e --- /dev/null +++ b/source/util/test/queueTest.cpp @@ -0,0 +1,29 @@ +/** + * @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 + +#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