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

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

This commit is contained in:
Hongze Cheng 2022-01-05 18:23:43 +08:00
parent 381cd7e4d7 2da55e1c71
commit 14ee6732d1
34 changed files with 1278 additions and 544 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/dnode/mnode/sdb/sdb.h
View File

@ -94,6 +94,7 @@ typedef struct SSdbRaw SSdbRaw;
typedef struct SSdbRow SSdbRow;
typedef enum { SDB_KEY_BINARY = 1, SDB_KEY_INT32 = 2, SDB_KEY_INT64 = 3 } EKeyType;
typedef enum {
SDB_STATUS_INIT = 0,
SDB_STATUS_CREATING = 1,
SDB_STATUS_UPDATING = 2,
SDB_STATUS_DROPPING = 3,

19
include/libs/catalog/catalog.h
View File

@ -50,6 +50,7 @@ typedef struct SCatalogCfg {
uint32_t maxDBCacheNum;
} SCatalogCfg;
int32_t catalogInit(SCatalogCfg *cfg);
/**
@ -87,15 +88,28 @@ int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDB
*/
int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
/**
* Get a super table's meta data.
* @param pCatalog (input, got with catalogGetHandle)
* @param pTransporter (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
* @param pTableName (input, table name, NOT including db name)
* @param pTableMeta(output, table meta data, NEED to free it by calller)
* @return error code
*/
int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
/**
* Force renew a table's local cached meta data.
* @param pCatalog (input, got with catalogGetHandle)
* @param pTransporter (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
* @param pTableName (input, table name, NOT including db name)
* @param isSTable (input, is super table or not, 1:supposed to be stable, 0: supposed not to be stable, -1:not sure)
* @return error code
*/
int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName);
int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable);
/**
* Force renew a table's local cached meta data and get the new one.
@ -104,9 +118,10 @@ int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void * pTransporter, co
* @param pMgmtEps (input, mnode EPs)
* @param pTableName (input, table name, NOT including db name)
* @param pTableMeta(output, table meta data, NEED to free it by calller)
* @param isSTable (input, is super table or not, 1:supposed to be stable, 0: supposed not to be stable, -1:not sure)
* @return error code
*/
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable);
/**

2
include/libs/parser/parser.h
View File

@ -43,7 +43,7 @@ int32_t qParseQuerySql(SParseContext* pContext, SQueryNode** pQuery);
bool qIsDdlQuery(const SQueryNode* pQuery);
void qDestroyQuery(SQueryNode* pQuery);
void qDestroyQuery(SQueryNode* pQueryNode);
/**
* Convert a normal sql statement to only query tags information to enable that the subscribe client can be aware quickly of the true vgroup ids that

16
source/client/src/clientEnv.c
View File

@ -180,6 +180,14 @@ void* createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t ty
return pRequest;
}
static void doFreeReqResultInfo(SReqResultInfo* pResInfo) {
tfree(pResInfo->pRspMsg);
tfree(pResInfo->length);
tfree(pResInfo->row);
tfree(pResInfo->pCol);
tfree(pResInfo->fields);
}
static void doDestroyRequest(void* p) {
assert(p != NULL);
SRequestObj* pRequest = (SRequestObj*)p;
@ -190,7 +198,7 @@ static void doDestroyRequest(void* p) {
tfree(pRequest->sqlstr);
tfree(pRequest->pInfo);
tfree(pRequest->body.resInfo.pRspMsg);
doFreeReqResultInfo(&pRequest->body.resInfo);
deregisterRequest(pRequest);
tfree(pRequest);
@ -415,7 +423,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
*+------------+-----+-----------+---------------+
*| uid|localIp| PId | timestamp | serial number |
*+------------+-----+-----------+---------------+
*| 16bit |12bit|20bit |16bit |
*| 12bit |12bit|24bit |16bit |
*+------------+-----+-----------+---------------+
* @return
*/
@ -435,11 +443,11 @@ uint64_t generateRequestId() {
}
}
int64_t ts = taosGetTimestampUs();
int64_t ts = taosGetTimestampMs();
uint64_t pid = taosGetPId();
int32_t val = atomic_add_fetch_32(&requestSerialId, 1);
uint64_t id = ((hashId & 0xFFFF) << 48) | ((pid & 0x0FFF) << 36) | ((ts & 0xFFFFF) << 16) | (val & 0xFFFF);
uint64_t id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
return id;
}

28
source/client/src/clientImpl.c
View File

@ -192,13 +192,12 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
}
tsem_wait(&pRequest->body.rspSem);
destroySendMsgInfo(pSendMsg);
return TSDB_CODE_SUCCESS;
}
int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQuery, SQueryDag** pDag) {
pRequest->type = pQuery->type;
return qCreateQueryDag(pQuery, pDag, pRequest->requestId);
int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag) {
pRequest->type = pQueryNode->type;
return qCreateQueryDag(pQueryNode, pDag, pRequest->requestId);
}
int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, void** pJob) {
@ -364,8 +363,6 @@ STscObj* taosConnectImpl(const char *ip, const char *user, const char *auth, con
asyncSendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
destroySendMsgInfo(body);
if (pRequest->code != TSDB_CODE_SUCCESS) {
const char *errorMsg = (pRequest->code == TSDB_CODE_RPC_FQDN_ERROR) ? taos_errstr(pRequest) : tstrerror(terrno);
printf("failed to connect to server, reason: %s\n\n", errorMsg);
@ -456,17 +453,21 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
taosReleaseRef(clientReqRefPool, pSendInfo->requestObjRefId);
}
SDataBuf buf = {.len = pMsg->contLen};
buf.pData = calloc(1, pMsg->contLen);
if (buf.pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
pMsg->code = TSDB_CODE_OUT_OF_MEMORY;
} else {
memcpy(buf.pData, pMsg->pCont, pMsg->contLen);
SDataBuf buf = {.len = pMsg->contLen, .pData = NULL};
if (pMsg->contLen > 0) {
buf.pData = calloc(1, pMsg->contLen);
if (buf.pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
pMsg->code = TSDB_CODE_OUT_OF_MEMORY;
} else {
memcpy(buf.pData, pMsg->pCont, pMsg->contLen);
}
}
pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);
rpcFreeCont(pMsg->pCont);
destroySendMsgInfo(pSendInfo);
}
TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port) {
@ -539,7 +540,6 @@ void* doFetchRow(SRequestObj* pRequest) {
asyncSendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
destroySendMsgInfo(body);
pResultInfo->current = 0;
if (pResultInfo->numOfRows <= pResultInfo->current) {

7
source/client/src/clientMsgHandler.c
View File

@ -30,6 +30,7 @@ int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
setErrno(pRequest, code);
free(pMsg->pData);
sem_post(&pRequest->body.rspSem);
return code;
}
@ -37,6 +38,7 @@ int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
free(pMsg->pData);
setErrno(pRequest, code);
sem_post(&pRequest->body.rspSem);
return code;
@ -73,6 +75,7 @@ int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
tscDebug("0x%" PRIx64 " clusterId:%" PRId64 ", totalConn:%" PRId64, pRequest->requestId, pConnect->clusterId,
pTscObj->pAppInfo->numOfConns);
free(pMsg->pData);
sem_post(&pRequest->body.rspSem);
return 0;
}
@ -238,6 +241,7 @@ int32_t processRetrieveVndRsp(void* param, const SDataBuf* pMsg, int32_t code) {
int32_t processCreateDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
// todo rsp with the vnode id list
SRequestObj* pRequest = param;
free(pMsg->pData);
tsem_post(&pRequest->body.rspSem);
}
@ -245,6 +249,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
free(pMsg->pData);
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
@ -258,6 +263,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
tNameGetDbName(&name, db);
setConnectionDB(pRequest->pTscObj, db);
free(pMsg->pData);
tsem_post(&pRequest->body.rspSem);
return 0;
}
@ -266,6 +272,7 @@ int32_t processCreateTableRsp(void* param, const SDataBuf* pMsg, int32_t code) {
assert(pMsg != NULL && param != NULL);
SRequestObj* pRequest = param;
free(pMsg->pData);
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);

173
source/client/test/clientTests.cpp
View File

@ -112,6 +112,7 @@ TEST(testCase, show_user_Test) {
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
@ -130,7 +131,7 @@ TEST(testCase, drop_user_Test) {
TEST(testCase, show_db_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "show databases");
TAOS_ROW pRow = NULL;
@ -170,62 +171,62 @@ TEST(testCase, create_db_Test) {
}
taos_close(pConn);
}
//
//TEST(testCase, create_dnode_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create dnode abc1 port 7000");
// if (taos_errno(pRes) != 0) {
// printf("error in create dnode, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create dnode 1.1.1.1 port 9000");
// if (taos_errno(pRes) != 0) {
// printf("failed to create dnode, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// taos_close(pConn);
//}
//
//TEST(testCase, drop_dnode_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "drop dnode 2");
// if (taos_errno(pRes) != 0) {
// printf("error in drop dnode, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, use_db_test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in use db, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_close(pConn);
//}
TEST(testCase, create_dnode_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "create dnode abc1 port 7000");
if (taos_errno(pRes) != 0) {
printf("error in create dnode, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create dnode 1.1.1.1 port 9000");
if (taos_errno(pRes) != 0) {
printf("failed to create dnode, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, drop_dnode_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "drop dnode 2");
if (taos_errno(pRes) != 0) {
printf("error in drop dnode, reason:%s\n", taos_errstr(pRes));
}
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
ASSERT_TRUE(pFields == NULL);
int32_t numOfFields = taos_num_fields(pRes);
ASSERT_EQ(numOfFields, 0);
taos_free_result(pRes);
taos_close(pConn);
}
TEST(testCase, use_db_test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
if (taos_errno(pRes) != 0) {
printf("error in use db, reason:%s\n", taos_errstr(pRes));
}
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
ASSERT_TRUE(pFields == NULL);
int32_t numOfFields = taos_num_fields(pRes);
ASSERT_EQ(numOfFields, 0);
taos_close(pConn);
}
//TEST(testCase, drop_db_test) {
//// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
@ -280,19 +281,19 @@ TEST(testCase, create_db_Test) {
taos_close(pConn);
}
//TEST(testCase, create_table_Test) {
// // TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// // assert(pConn != NULL);
// //
// // TAOS_RES* pRes = taos_query(pConn, "use abc1");
// // taos_free_result(pRes);
// //
// // pRes = taos_query(pConn, "create table tm0(ts timestamp, k int)");
// // taos_free_result(pRes);
// //
// // taos_close(pConn);
//}
//
TEST(testCase, create_table_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
taos_free_result(pRes);
pRes = taos_query(pConn, "create table tm0(ts timestamp, k int)");
taos_free_result(pRes);
taos_close(pConn);
}
//TEST(testCase, create_ctable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
@ -303,12 +304,12 @@ TEST(testCase, create_db_Test) {
// }
// taos_free_result(pRes);
//
//// pRes = taos_query(pConn, "create table tm0 using st1 tags(1)");
//// if (taos_errno(pRes) != 0) {
//// printf("failed to create child table tm0, reason:%s\n", taos_errstr(pRes));
//// }
////
//// taos_free_result(pRes);
// pRes = taos_query(pConn, "create table tm0 using st1 tags(1)");
// if (taos_errno(pRes) != 0) {
// printf("failed to create child table tm0, reason:%s\n", taos_errstr(pRes));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
@ -342,7 +343,7 @@ TEST(testCase, create_db_Test) {
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, show_vgroup_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
@ -375,7 +376,7 @@ TEST(testCase, create_db_Test) {
//
// taos_close(pConn);
//}
//
//TEST(testCase, drop_stable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
@ -468,6 +469,13 @@ TEST(testCase, create_multiple_tables) {
ASSERT_NE(pConn, nullptr);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
if (taos_errno(pRes) != 0) {
printf("failed to use db, reason:%s", taos_errstr(pRes));
taos_free_result(pRes);
taos_close(pConn);
return;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create table t_2 using st1 tags(1)");
@ -513,14 +521,17 @@ TEST(testCase, create_multiple_tables) {
TEST(testCase, generated_request_id_test) {
SHashObj *phash = taosHashInit(10000, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
for(int32_t i = 0; i < 1000000; ++i) {
for(int32_t i = 0; i < 50000000; ++i) {
uint64_t v = generateRequestId();
void* result = taosHashGet(phash, &v, sizeof(v));
ASSERT_EQ(result, nullptr);
if (result != nullptr) {
printf("0x%"PRIx64", index:%d\n", v, i);
}
assert(result == nullptr);
taosHashPut(phash, &v, sizeof(v), NULL, 0);
}
taosHashClear(phash);
taosHashCleanup(phash);
}
//TEST(testCase, projection_query_tables) {

4
source/dnode/mgmt/impl/test/profile/profile.cpp
View File

@ -17,15 +17,15 @@ class DndTestProfile : public ::testing::Test {
static void TearDownTestSuite() { test.Cleanup(); }
static Testbase test;
static int32_t connId;
public:
void SetUp() override {}
void TearDown() override {}
int32_t connId;
};
Testbase DndTestProfile::test;
int32_t DndTestProfile::connId;
TEST_F(DndTestProfile, 01_ConnectMsg) {
int32_t contLen = sizeof(SConnectMsg);

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

@ -64,10 +64,10 @@ typedef enum {
TRN_STAGE_PREPARE = 0,
TRN_STAGE_REDO_LOG = 1,
TRN_STAGE_REDO_ACTION = 2,
TRN_STAGE_UNDO_LOG = 3,
TRN_STAGE_UNDO_ACTION = 4,
TRN_STAGE_COMMIT_LOG = 5,
TRN_STAGE_COMMIT = 6,
TRN_STAGE_COMMIT_LOG = 3,
TRN_STAGE_COMMIT = 4,
TRN_STAGE_UNDO_ACTION = 5,
TRN_STAGE_UNDO_LOG = 6,
TRN_STAGE_ROLLBACK = 7,
TRN_STAGE_FINISHED = 8
} ETrnStage;

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

@ -28,6 +28,7 @@ typedef struct {
int8_t msgSent;
int8_t msgReceived;
int32_t errCode;
int32_t acceptableCode;
int32_t contLen;
void *pCont;
} STransAction;

123
source/dnode/mnode/impl/src/mndBnode.c
View File

@ -26,7 +26,7 @@ static SSdbRaw *mndBnodeActionEncode(SBnodeObj *pObj);
static SSdbRow *mndBnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndBnodeActionInsert(SSdb *pSdb, SBnodeObj *pObj);
static int32_t mndBnodeActionDelete(SSdb *pSdb, SBnodeObj *pObj);
static int32_t mndBnodeActionUpdate(SSdb *pSdb, SBnodeObj *pOldBnode, SBnodeObj *pNewBnode);
static int32_t mndBnodeActionUpdate(SSdb *pSdb, SBnodeObj *pOld, SBnodeObj *pNew);
static int32_t mndProcessCreateBnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessDropBnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessCreateBnodeRsp(SMnodeMsg *pMsg);
@ -59,9 +59,8 @@ int32_t mndInitBnode(SMnode *pMnode) {
void mndCleanupBnode(SMnode *pMnode) {}
static SBnodeObj *mndAcquireBnode(SMnode *pMnode, int32_t bnodeId) {
SSdb *pSdb = pMnode->pSdb;
SBnodeObj *pObj = sdbAcquire(pSdb, SDB_BNODE, &bnodeId);
if (pObj == NULL) {
SBnodeObj *pObj = sdbAcquire(pMnode->pSdb, SDB_BNODE, &bnodeId);
if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
terrno = TSDB_CODE_MND_BNODE_NOT_EXIST;
}
return pObj;
@ -155,9 +154,9 @@ static int32_t mndBnodeActionDelete(SSdb *pSdb, SBnodeObj *pObj) {
return 0;
}
static int32_t mndBnodeActionUpdate(SSdb *pSdb, SBnodeObj *pOldBnode, SBnodeObj *pNewBnode) {
mTrace("bnode:%d, perform update action, old_row:%p new_row:%p", pOldBnode->id, pOldBnode, pNewBnode);
pOldBnode->updateTime = pNewBnode->updateTime;
static int32_t mndBnodeActionUpdate(SSdb *pSdb, SBnodeObj *pOld, SBnodeObj *pNew) {
mTrace("bnode:%d, perform update action, old_row:%p new_row:%p", pOld->id, pOld, pNew);
pOld->updateTime = pNew->updateTime;
return 0;
}
@ -169,6 +168,14 @@ static int32_t mndSetCreateBnodeRedoLogs(STrans *pTrans, SBnodeObj *pObj) {
return 0;
}
static int32_t mndSetCreateBnodeUndoLogs(STrans *pTrans, SBnodeObj *pObj) {
SSdbRaw *pUndoRaw = mndBnodeActionEncode(pObj);
if (pUndoRaw == NULL) return -1;
if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) return -1;
if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED) != 0) return -1;
return 0;
}
static int32_t mndSetCreateBnodeCommitLogs(STrans *pTrans, SBnodeObj *pObj) {
SSdbRaw *pCommitRaw = mndBnodeActionEncode(pObj);
if (pCommitRaw == NULL) return -1;
@ -190,6 +197,7 @@ static int32_t mndSetCreateBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
action.pCont = pMsg;
action.contLen = sizeof(SDCreateBnodeReq);
action.msgType = TDMT_DND_CREATE_BNODE;
action.acceptableCode = TSDB_CODE_DND_BNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
@ -199,39 +207,47 @@ static int32_t mndSetCreateBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
return 0;
}
static int32_t mndSetCreateBnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, SBnodeObj *pObj) {
SDDropBnodeReq *pMsg = malloc(sizeof(SDDropBnodeReq));
if (pMsg == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pMsg->dnodeId = htonl(pDnode->id);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pMsg;
action.contLen = sizeof(SDDropBnodeReq);
action.msgType = TDMT_DND_DROP_BNODE;
action.acceptableCode = TSDB_CODE_DND_BNODE_NOT_DEPLOYED;
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
free(pMsg);
return -1;
}
return 0;
}
static int32_t mndCreateBnode(SMnode *pMnode, SMnodeMsg *pMsg, SDnodeObj *pDnode, SMCreateBnodeReq *pCreate) {
int32_t code = -1;
SBnodeObj bnodeObj = {0};
bnodeObj.id = pDnode->id;
bnodeObj.createdTime = taosGetTimestampMs();
bnodeObj.updateTime = bnodeObj.createdTime;
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("bnode:%d, failed to create since %s", pCreate->dnodeId, terrstr());
goto CREATE_BNODE_OVER;
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
if (pTrans == NULL) goto CREATE_BNODE_OVER;
mDebug("trans:%d, used to create bnode:%d", pTrans->id, pCreate->dnodeId);
if (mndSetCreateBnodeRedoLogs(pTrans, &bnodeObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto CREATE_BNODE_OVER;
}
if (mndSetCreateBnodeCommitLogs(pTrans, &bnodeObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto CREATE_BNODE_OVER;
}
if (mndSetCreateBnodeRedoActions(pTrans, pDnode, &bnodeObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto CREATE_BNODE_OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto CREATE_BNODE_OVER;
}
if (mndSetCreateBnodeRedoLogs(pTrans, &bnodeObj) != 0) goto CREATE_BNODE_OVER;
if (mndSetCreateBnodeUndoLogs(pTrans, &bnodeObj) != 0) goto CREATE_BNODE_OVER;
if (mndSetCreateBnodeCommitLogs(pTrans, &bnodeObj) != 0) goto CREATE_BNODE_OVER;
if (mndSetCreateBnodeRedoActions(pTrans, pDnode, &bnodeObj) != 0) goto CREATE_BNODE_OVER;
if (mndSetCreateBnodeUndoActions(pTrans, pDnode, &bnodeObj) != 0) goto CREATE_BNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto CREATE_BNODE_OVER;
code = 0;
@ -251,8 +267,12 @@ static int32_t mndProcessCreateBnodeReq(SMnodeMsg *pMsg) {
SBnodeObj *pObj = mndAcquireBnode(pMnode, pCreate->dnodeId);
if (pObj != NULL) {
mError("bnode:%d, bnode already exist", pObj->id);
terrno = TSDB_CODE_MND_BNODE_ALREADY_EXIST;
mndReleaseBnode(pMnode, pObj);
return -1;
} else if (terrno != TSDB_CODE_MND_BNODE_NOT_EXIST) {
mError("bnode:%d, failed to create bnode since %s", pCreate->dnodeId, terrstr());
return -1;
}
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pCreate->dnodeId);
@ -302,6 +322,7 @@ static int32_t mndSetDropBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SBn
action.pCont = pMsg;
action.contLen = sizeof(SDDropBnodeReq);
action.msgType = TDMT_DND_DROP_BNODE;
action.acceptableCode = TSDB_CODE_DND_BNODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
@ -313,33 +334,15 @@ static int32_t mndSetDropBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SBn
static int32_t mndDropBnode(SMnode *pMnode, SMnodeMsg *pMsg, SBnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("bnode:%d, failed to drop since %s", pObj->id, terrstr());
goto DROP_BNODE_OVER;
}
if (pTrans == NULL) goto DROP_BNODE_OVER;
mDebug("trans:%d, used to drop bnode:%d", pTrans->id, pObj->id);
if (mndSetDropBnodeRedoLogs(pTrans, pObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto DROP_BNODE_OVER;
}
if (mndSetDropBnodeCommitLogs(pTrans, pObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto DROP_BNODE_OVER;
}
if (mndSetDropBnodeRedoActions(pTrans, pObj->pDnode, pObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto DROP_BNODE_OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto DROP_BNODE_OVER;
}
if (mndSetDropBnodeRedoLogs(pTrans, pObj) != 0) goto DROP_BNODE_OVER;
if (mndSetDropBnodeCommitLogs(pTrans, pObj) != 0) goto DROP_BNODE_OVER;
if (mndSetDropBnodeRedoActions(pTrans, pObj->pDnode, pObj) != 0) goto DROP_BNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto DROP_BNODE_OVER;
code = 0;
@ -363,18 +366,18 @@ static int32_t mndProcessDropBnodeReq(SMnodeMsg *pMsg) {
SBnodeObj *pObj = mndAcquireBnode(pMnode, pDrop->dnodeId);
if (pObj == NULL) {
mError("bnode:%d, not exist", pDrop->dnodeId);
terrno = TSDB_CODE_MND_BNODE_NOT_EXIST;
mError("bnode:%d, failed to drop since %s", pDrop->dnodeId, terrstr());
return -1;
}
int32_t code = mndDropBnode(pMnode, pMsg, pObj);
if (code != 0) {
sdbRelease(pMnode->pSdb, pObj);
mError("bnode:%d, failed to drop since %s", pMnode->dnodeId, terrstr());
return -1;
}
sdbRelease(pMnode->pSdb, pMnode);
sdbRelease(pMnode->pSdb, pObj);
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}

123
source/dnode/mnode/impl/src/mndQnode.c
View File

@ -26,7 +26,7 @@ static SSdbRaw *mndQnodeActionEncode(SQnodeObj *pObj);
static SSdbRow *mndQnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndQnodeActionInsert(SSdb *pSdb, SQnodeObj *pObj);
static int32_t mndQnodeActionDelete(SSdb *pSdb, SQnodeObj *pObj);
static int32_t mndQnodeActionUpdate(SSdb *pSdb, SQnodeObj *pOldQnode, SQnodeObj *pNewQnode);
static int32_t mndQnodeActionUpdate(SSdb *pSdb, SQnodeObj *pOld, SQnodeObj *pNew);
static int32_t mndProcessCreateQnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessDropQnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessCreateQnodeRsp(SMnodeMsg *pMsg);
@ -59,9 +59,8 @@ int32_t mndInitQnode(SMnode *pMnode) {
void mndCleanupQnode(SMnode *pMnode) {}
static SQnodeObj *mndAcquireQnode(SMnode *pMnode, int32_t qnodeId) {
SSdb *pSdb = pMnode->pSdb;
SQnodeObj *pObj = sdbAcquire(pSdb, SDB_QNODE, &qnodeId);
if (pObj == NULL) {
SQnodeObj *pObj = sdbAcquire(pMnode->pSdb, SDB_QNODE, &qnodeId);
if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
terrno = TSDB_CODE_MND_QNODE_NOT_EXIST;
}
return pObj;
@ -155,9 +154,9 @@ static int32_t mndQnodeActionDelete(SSdb *pSdb, SQnodeObj *pObj) {
return 0;
}
static int32_t mndQnodeActionUpdate(SSdb *pSdb, SQnodeObj *pOldQnode, SQnodeObj *pNewQnode) {
mTrace("qnode:%d, perform update action, old_row:%p new_row:%p", pOldQnode->id, pOldQnode, pNewQnode);
pOldQnode->updateTime = pNewQnode->updateTime;
static int32_t mndQnodeActionUpdate(SSdb *pSdb, SQnodeObj *pOld, SQnodeObj *pNew) {
mTrace("qnode:%d, perform update action, old_row:%p new_row:%p", pOld->id, pOld, pNew);
pOld->updateTime = pNew->updateTime;
return 0;
}
@ -169,6 +168,14 @@ static int32_t mndSetCreateQnodeRedoLogs(STrans *pTrans, SQnodeObj *pObj) {
return 0;
}
static int32_t mndSetCreateQnodeUndoLogs(STrans *pTrans, SQnodeObj *pObj) {
SSdbRaw *pUndoRaw = mndQnodeActionEncode(pObj);
if (pUndoRaw == NULL) return -1;
if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) return -1;
if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED) != 0) return -1;
return 0;
}
static int32_t mndSetCreateQnodeCommitLogs(STrans *pTrans, SQnodeObj *pObj) {
SSdbRaw *pCommitRaw = mndQnodeActionEncode(pObj);
if (pCommitRaw == NULL) return -1;
@ -190,6 +197,7 @@ static int32_t mndSetCreateQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
action.pCont = pMsg;
action.contLen = sizeof(SDCreateQnodeReq);
action.msgType = TDMT_DND_CREATE_QNODE;
action.acceptableCode = TSDB_CODE_DND_QNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
@ -199,39 +207,47 @@ static int32_t mndSetCreateQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
return 0;
}
static int32_t mndSetCreateQnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, SQnodeObj *pObj) {
SDDropQnodeReq *pMsg = malloc(sizeof(SDDropQnodeReq));
if (pMsg == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pMsg->dnodeId = htonl(pDnode->id);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pMsg;
action.contLen = sizeof(SDDropQnodeReq);
action.msgType = TDMT_DND_DROP_QNODE;
action.acceptableCode = TSDB_CODE_DND_QNODE_NOT_DEPLOYED;
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
free(pMsg);
return -1;
}
return 0;
}
static int32_t mndCreateQnode(SMnode *pMnode, SMnodeMsg *pMsg, SDnodeObj *pDnode, SMCreateQnodeReq *pCreate) {
int32_t code = -1;
SQnodeObj qnodeObj = {0};
qnodeObj.id = pDnode->id;
qnodeObj.createdTime = taosGetTimestampMs();
qnodeObj.updateTime = qnodeObj.createdTime;
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("qnode:%d, failed to create since %s", pCreate->dnodeId, terrstr());
goto CREATE_QNODE_OVER;
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
if (pTrans == NULL) goto CREATE_QNODE_OVER;
mDebug("trans:%d, used to create qnode:%d", pTrans->id, pCreate->dnodeId);
if (mndSetCreateQnodeRedoLogs(pTrans, &qnodeObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto CREATE_QNODE_OVER;
}
if (mndSetCreateQnodeCommitLogs(pTrans, &qnodeObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto CREATE_QNODE_OVER;
}
if (mndSetCreateQnodeRedoActions(pTrans, pDnode, &qnodeObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto CREATE_QNODE_OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto CREATE_QNODE_OVER;
}
if (mndSetCreateQnodeRedoLogs(pTrans, &qnodeObj) != 0) goto CREATE_QNODE_OVER;
if (mndSetCreateQnodeUndoLogs(pTrans, &qnodeObj) != 0) goto CREATE_QNODE_OVER;
if (mndSetCreateQnodeCommitLogs(pTrans, &qnodeObj) != 0) goto CREATE_QNODE_OVER;
if (mndSetCreateQnodeRedoActions(pTrans, pDnode, &qnodeObj) != 0) goto CREATE_QNODE_OVER;
if (mndSetCreateQnodeUndoActions(pTrans, pDnode, &qnodeObj) != 0) goto CREATE_QNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto CREATE_QNODE_OVER;
code = 0;
@ -251,8 +267,12 @@ static int32_t mndProcessCreateQnodeReq(SMnodeMsg *pMsg) {
SQnodeObj *pObj = mndAcquireQnode(pMnode, pCreate->dnodeId);
if (pObj != NULL) {
mError("qnode:%d, qnode already exist", pObj->id);
terrno = TSDB_CODE_MND_QNODE_ALREADY_EXIST;
mndReleaseQnode(pMnode, pObj);
return -1;
} else if (terrno != TSDB_CODE_MND_QNODE_NOT_EXIST) {
mError("qnode:%d, failed to create qnode since %s", pCreate->dnodeId, terrstr());
return -1;
}
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pCreate->dnodeId);
@ -302,6 +322,7 @@ static int32_t mndSetDropQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQn
action.pCont = pMsg;
action.contLen = sizeof(SDDropQnodeReq);
action.msgType = TDMT_DND_DROP_QNODE;
action.acceptableCode = TSDB_CODE_DND_QNODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
@ -313,33 +334,15 @@ static int32_t mndSetDropQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQn
static int32_t mndDropQnode(SMnode *pMnode, SMnodeMsg *pMsg, SQnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("qnode:%d, failed to drop since %s", pObj->id, terrstr());
goto DROP_QNODE_OVER;
}
if (pTrans == NULL) goto DROP_QNODE_OVER;
mDebug("trans:%d, used to drop qnode:%d", pTrans->id, pObj->id);
if (mndSetDropQnodeRedoLogs(pTrans, pObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto DROP_QNODE_OVER;
}
if (mndSetDropQnodeCommitLogs(pTrans, pObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto DROP_QNODE_OVER;
}
if (mndSetDropQnodeRedoActions(pTrans, pObj->pDnode, pObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto DROP_QNODE_OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto DROP_QNODE_OVER;
}
if (mndSetDropQnodeRedoLogs(pTrans, pObj) != 0) goto DROP_QNODE_OVER;
if (mndSetDropQnodeCommitLogs(pTrans, pObj) != 0) goto DROP_QNODE_OVER;
if (mndSetDropQnodeRedoActions(pTrans, pObj->pDnode, pObj) != 0) goto DROP_QNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto DROP_QNODE_OVER;
code = 0;
@ -363,18 +366,18 @@ static int32_t mndProcessDropQnodeReq(SMnodeMsg *pMsg) {
SQnodeObj *pObj = mndAcquireQnode(pMnode, pDrop->dnodeId);
if (pObj == NULL) {
mError("qnode:%d, not exist", pDrop->dnodeId);
terrno = TSDB_CODE_MND_QNODE_NOT_EXIST;
mError("qnode:%d, failed to drop since %s", pDrop->dnodeId, terrstr());
return -1;
}
int32_t code = mndDropQnode(pMnode, pMsg, pObj);
if (code != 0) {
sdbRelease(pMnode->pSdb, pObj);
mError("qnode:%d, failed to drop since %s", pMnode->dnodeId, terrstr());
return -1;
}
sdbRelease(pMnode->pSdb, pMnode);
sdbRelease(pMnode->pSdb, pObj);
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}

121
source/dnode/mnode/impl/src/mndSnode.c
View File

@ -26,7 +26,7 @@ static SSdbRaw *mndSnodeActionEncode(SSnodeObj *pObj);
static SSdbRow *mndSnodeActionDecode(SSdbRaw *pRaw);
static int32_t mndSnodeActionInsert(SSdb *pSdb, SSnodeObj *pObj);
static int32_t mndSnodeActionDelete(SSdb *pSdb, SSnodeObj *pObj);
static int32_t mndSnodeActionUpdate(SSdb *pSdb, SSnodeObj *pOldSnode, SSnodeObj *pNewSnode);
static int32_t mndSnodeActionUpdate(SSdb *pSdb, SSnodeObj *pOld, SSnodeObj *pNew);
static int32_t mndProcessCreateSnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessDropSnodeReq(SMnodeMsg *pMsg);
static int32_t mndProcessCreateSnodeRsp(SMnodeMsg *pMsg);
@ -59,9 +59,8 @@ int32_t mndInitSnode(SMnode *pMnode) {
void mndCleanupSnode(SMnode *pMnode) {}
static SSnodeObj *mndAcquireSnode(SMnode *pMnode, int32_t snodeId) {
SSdb *pSdb = pMnode->pSdb;
SSnodeObj *pObj = sdbAcquire(pSdb, SDB_SNODE, &snodeId);
if (pObj == NULL) {
SSnodeObj *pObj = sdbAcquire(pMnode->pSdb, SDB_SNODE, &snodeId);
if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
terrno = TSDB_CODE_MND_SNODE_NOT_EXIST;
}
return pObj;
@ -155,9 +154,9 @@ static int32_t mndSnodeActionDelete(SSdb *pSdb, SSnodeObj *pObj) {
return 0;
}
static int32_t mndSnodeActionUpdate(SSdb *pSdb, SSnodeObj *pOldSnode, SSnodeObj *pNewSnode) {
mTrace("snode:%d, perform update action, old_row:%p new_row:%p", pOldSnode->id, pOldSnode, pNewSnode);
pOldSnode->updateTime = pNewSnode->updateTime;
static int32_t mndSnodeActionUpdate(SSdb *pSdb, SSnodeObj *pOld, SSnodeObj *pNew) {
mTrace("snode:%d, perform update action, old_row:%p new_row:%p", pOld->id, pOld, pNew);
pOld->updateTime = pNew->updateTime;
return 0;
}
@ -169,6 +168,14 @@ static int32_t mndSetCreateSnodeRedoLogs(STrans *pTrans, SSnodeObj *pObj) {
return 0;
}
static int32_t mndSetCreateSnodeUndoLogs(STrans *pTrans, SSnodeObj *pObj) {
SSdbRaw *pUndoRaw = mndSnodeActionEncode(pObj);
if (pUndoRaw == NULL) return -1;
if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) return -1;
if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED) != 0) return -1;
return 0;
}
static int32_t mndSetCreateSnodeCommitLogs(STrans *pTrans, SSnodeObj *pObj) {
SSdbRaw *pCommitRaw = mndSnodeActionEncode(pObj);
if (pCommitRaw == NULL) return -1;
@ -190,6 +197,7 @@ static int32_t mndSetCreateSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
action.pCont = pMsg;
action.contLen = sizeof(SDCreateSnodeReq);
action.msgType = TDMT_DND_CREATE_SNODE;
action.acceptableCode = TSDB_CODE_DND_SNODE_ALREADY_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
@ -199,39 +207,48 @@ static int32_t mndSetCreateSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, S
return 0;
}
static int32_t mndSetCreateSnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, SSnodeObj *pObj) {
SDDropSnodeReq *pMsg = malloc(sizeof(SDDropSnodeReq));
if (pMsg == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pMsg->dnodeId = htonl(pDnode->id);
STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode);
action.pCont = pMsg;
action.contLen = sizeof(SDDropSnodeReq);
action.msgType = TDMT_DND_DROP_SNODE;
action.acceptableCode = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
free(pMsg);
return -1;
}
return 0;
}
static int32_t mndCreateSnode(SMnode *pMnode, SMnodeMsg *pMsg, SDnodeObj *pDnode, SMCreateSnodeReq *pCreate) {
int32_t code = -1;
SSnodeObj snodeObj = {0};
snodeObj.id = pDnode->id;
snodeObj.createdTime = taosGetTimestampMs();
snodeObj.updateTime = snodeObj.createdTime;
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("snode:%d, failed to create since %s", pCreate->dnodeId, terrstr());
goto CREATE_SNODE_OVER;
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
if (pTrans == NULL) goto CREATE_SNODE_OVER;
mDebug("trans:%d, used to create snode:%d", pTrans->id, pCreate->dnodeId);
if (mndSetCreateSnodeRedoLogs(pTrans, &snodeObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto CREATE_SNODE_OVER;
}
if (mndSetCreateSnodeCommitLogs(pTrans, &snodeObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto CREATE_SNODE_OVER;
}
if (mndSetCreateSnodeRedoActions(pTrans, pDnode, &snodeObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto CREATE_SNODE_OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto CREATE_SNODE_OVER;
}
if (mndSetCreateSnodeRedoLogs(pTrans, &snodeObj) != 0) goto CREATE_SNODE_OVER;
if (mndSetCreateSnodeUndoLogs(pTrans, &snodeObj) != 0) goto CREATE_SNODE_OVER;
if (mndSetCreateSnodeCommitLogs(pTrans, &snodeObj) != 0) goto CREATE_SNODE_OVER;
if (mndSetCreateSnodeRedoActions(pTrans, pDnode, &snodeObj) != 0) goto CREATE_SNODE_OVER;
if (mndSetCreateSnodeUndoActions(pTrans, pDnode, &snodeObj) != 0) goto CREATE_SNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto CREATE_SNODE_OVER;
code = 0;
@ -251,8 +268,12 @@ static int32_t mndProcessCreateSnodeReq(SMnodeMsg *pMsg) {
SSnodeObj *pObj = mndAcquireSnode(pMnode, pCreate->dnodeId);
if (pObj != NULL) {
mError("snode:%d, snode already exist", pObj->id);
terrno = TSDB_CODE_MND_SNODE_ALREADY_EXIST;
mndReleaseSnode(pMnode, pObj);
return -1;
} else if (terrno != TSDB_CODE_MND_SNODE_NOT_EXIST) {
mError("snode:%d, failed to create snode since %s", pCreate->dnodeId, terrstr());
return -1;
}
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pCreate->dnodeId);
@ -302,6 +323,7 @@ static int32_t mndSetDropSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SSn
action.pCont = pMsg;
action.contLen = sizeof(SDDropSnodeReq);
action.msgType = TDMT_DND_DROP_SNODE;
action.acceptableCode = TSDB_CODE_DND_SNODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
free(pMsg);
@ -313,33 +335,16 @@ static int32_t mndSetDropSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SSn
static int32_t mndDropSnode(SMnode *pMnode, SMnodeMsg *pMsg, SSnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
if (pTrans == NULL) {
mError("snode:%d, failed to drop since %s", pObj->id, terrstr());
goto DROP_SNODE_OVER;
}
if (pTrans == NULL) goto DROP_SNODE_OVER;
mDebug("trans:%d, used to drop snode:%d", pTrans->id, pObj->id);
if (mndSetDropSnodeRedoLogs(pTrans, pObj) != 0) {
mError("trans:%d, failed to set redo log since %s", pTrans->id, terrstr());
goto DROP_SNODE_OVER;
}
if (mndSetDropSnodeCommitLogs(pTrans, pObj) != 0) {
mError("trans:%d, failed to set commit log since %s", pTrans->id, terrstr());
goto DROP_SNODE_OVER;
}
if (mndSetDropSnodeRedoActions(pTrans, pObj->pDnode, pObj) != 0) {
mError("trans:%d, failed to set redo actions since %s", pTrans->id, terrstr());
goto DROP_SNODE_OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
goto DROP_SNODE_OVER;
}
if (mndSetDropSnodeRedoLogs(pTrans, pObj) != 0) goto DROP_SNODE_OVER;
if (mndSetDropSnodeCommitLogs(pTrans, pObj) != 0) goto DROP_SNODE_OVER;
if (mndSetDropSnodeRedoActions(pTrans, pObj->pDnode, pObj) != 0) goto DROP_SNODE_OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto DROP_SNODE_OVER;
code = 0;
@ -363,18 +368,18 @@ static int32_t mndProcessDropSnodeReq(SMnodeMsg *pMsg) {
SSnodeObj *pObj = mndAcquireSnode(pMnode, pDrop->dnodeId);
if (pObj == NULL) {
mError("snode:%d, not exist", pDrop->dnodeId);
terrno = TSDB_CODE_MND_SNODE_NOT_EXIST;
mError("snode:%d, failed to drop since %s", pDrop->dnodeId, terrstr());
return -1;
}
int32_t code = mndDropSnode(pMnode, pMsg, pObj);
if (code != 0) {
sdbRelease(pMnode->pSdb, pObj);
mError("snode:%d, failed to drop since %s", pMnode->dnodeId, terrstr());
return -1;
}
sdbRelease(pMnode->pSdb, pMnode);
sdbRelease(pMnode->pSdb, pObj);
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}

44
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -143,6 +143,7 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), TRANS_ENCODE_OVER)
SDB_SET_INT16(pRaw, dataPos, pAction->msgType, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, TRANS_ENCODE_OVER)
}
@ -151,6 +152,7 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), TRANS_ENCODE_OVER)
SDB_SET_INT16(pRaw, dataPos, pAction->msgType, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pCont, pAction->contLen, TRANS_ENCODE_OVER)
}
@ -253,6 +255,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
for (int32_t i = 0; i < redoActionNum; ++i) {
SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), TRANS_DECODE_OVER);
SDB_GET_INT16(pRaw, dataPos, &action.msgType, TRANS_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, TRANS_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &action.contLen, TRANS_DECODE_OVER)
action.pCont = malloc(action.contLen);
if (action.pCont == NULL) goto TRANS_DECODE_OVER;
@ -264,6 +267,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
for (int32_t i = 0; i < undoActionNum; ++i) {
SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), TRANS_DECODE_OVER);
SDB_GET_INT16(pRaw, dataPos, &action.msgType, TRANS_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, TRANS_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &action.contLen, TRANS_DECODE_OVER)
action.pCont = malloc(action.contLen);
if (action.pCont == NULL) goto TRANS_DECODE_OVER;
@ -496,10 +500,31 @@ static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
}
static void mndTransSendRpcRsp(STrans *pTrans) {
if (pTrans->rpcHandle != NULL) {
mDebug("trans:%d, send rsp, ahandle:%p code:0x%x", pTrans->id, pTrans->rpcAHandle, pTrans->code & 0xFFFF);
bool sendRsp = false;
if (pTrans->stage == TRN_STAGE_FINISHED) {
sendRsp = true;
}
if (pTrans->policy == TRN_POLICY_ROLLBACK) {
if (pTrans->stage == TRN_STAGE_UNDO_LOG || pTrans->stage == TRN_STAGE_UNDO_ACTION ||
pTrans->stage == TRN_STAGE_ROLLBACK) {
sendRsp = true;
}
}
if (pTrans->policy == TRN_POLICY_RETRY) {
if (pTrans->stage == TRN_STAGE_REDO_ACTION && pTrans->failedTimes > 0) {
sendRsp = true;
}
}
if (sendRsp && pTrans->rpcHandle != NULL) {
mDebug("trans:%d, send rsp, code:0x%x stage:%d app:%p", pTrans->id, pTrans->code & 0xFFFF, pTrans->stage,
pTrans->rpcAHandle);
SRpcMsg rspMsg = {.handle = pTrans->rpcHandle, .code = pTrans->code, .ahandle = pTrans->rpcAHandle};
rpcSendResponse(&rspMsg);
pTrans->rpcHandle = NULL;
}
}
@ -542,7 +567,8 @@ void mndTransProcessRsp(SMnodeMsg *pMsg) {
pAction->errCode = pMsg->rpcMsg.code;
}
mDebug("trans:%d, action:%d response is received, code:0x%x", transId, action, pMsg->rpcMsg.code);
mDebug("trans:%d, action:%d response is received, code:0x%x, accept:0x%x", transId, action, pMsg->rpcMsg.code,
pAction->acceptableCode);
mndTransExecute(pMnode, pTrans);
HANDLE_ACTION_RSP_OVER:
@ -642,7 +668,7 @@ static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pA
if (pAction == NULL) continue;
if (pAction->msgSent && pAction->msgReceived) {
numOfReceived++;
if (pAction->errCode != 0) {
if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
errCode = pAction->errCode;
}
}
@ -690,7 +716,7 @@ static bool mndTransPerformRedoLogStage(SMnode *pMnode, STrans *pTrans) {
} else {
pTrans->code = terrno;
pTrans->stage = TRN_STAGE_UNDO_LOG;
mError("trans:%d, stage from redoLog to undoLog", pTrans->id);
mError("trans:%d, stage from redoLog to undoLog since %s", pTrans->id, terrstr());
}
return continueExec;
@ -764,7 +790,6 @@ static bool mndTransPerformCommitLogStage(SMnode *pMnode, STrans *pTrans) {
pTrans->failedTimes++;
mError("trans:%d, stage keep on commitLog since %s", pTrans->id, terrstr());
continueExec = false;
;
}
return continueExec;
@ -791,7 +816,7 @@ static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans) {
int32_t code = mndTransExecuteUndoActions(pMnode, pTrans);
if (code == 0) {
pTrans->stage = TRN_STAGE_REDO_LOG;
pTrans->stage = TRN_STAGE_UNDO_LOG;
mDebug("trans:%d, stage from undoAction to undoLog", pTrans->id);
continueExec = true;
} else if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) {
@ -814,7 +839,6 @@ static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans) {
pTrans->stage = TRN_STAGE_FINISHED;
mDebug("trans:%d, stage from rollback to finished", pTrans->id);
continueExec = true;
;
} else {
pTrans->failedTimes++;
mError("trans:%d, stage keep on rollback since %s", pTrans->id, terrstr());
@ -880,9 +904,7 @@ static void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
}
}
if (pTrans->stage == TRN_STAGE_FINISHED) {
mndTransSendRpcRsp(pTrans);
}
mndTransSendRpcRsp(pTrans);
}
static int32_t mndProcessTransMsg(SMnodeMsg *pMsg) {

168
source/dnode/mnode/impl/test/bnode/bnode.cpp
View File

@ -50,7 +50,18 @@ TEST_F(MndTestBnode, 01_Show_Bnode) {
EXPECT_EQ(test.GetShowRows(), 0);
}
TEST_F(MndTestBnode, 02_Create_Bnode_Invalid_Id) {
TEST_F(MndTestBnode, 02_Create_Bnode) {
{
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_DNODE_NOT_EXIST);
}
{
int32_t contLen = sizeof(SMCreateBnodeReq);
@ -63,11 +74,6 @@ TEST_F(MndTestBnode, 02_Create_Bnode_Invalid_Id) {
test.SendShowMetaMsg(TSDB_MGMT_TABLE_BNODE, "");
CHECK_META("show bnodes", 3);
CHECK_SCHEMA(0, TSDB_DATA_TYPE_SMALLINT, 2, "id");
CHECK_SCHEMA(1, TSDB_DATA_TYPE_BINARY, TSDB_EP_LEN + VARSTR_HEADER_SIZE, "endpoint");
CHECK_SCHEMA(2, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
test.SendShowRetrieveMsg();
EXPECT_EQ(test.GetShowRows(), 1);
@ -75,24 +81,21 @@ TEST_F(MndTestBnode, 02_Create_Bnode_Invalid_Id) {
CheckBinary("localhost:9018", TSDB_EP_LEN);
CheckTimestamp();
}
}
TEST_F(MndTestBnode, 03_Create_Bnode_Invalid_Id) {
{
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
pReq->dnodeId = htonl(1);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_DNODE_NOT_EXIST);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_BNODE_ALREADY_EXIST);
}
}
TEST_F(MndTestBnode, 04_Create_Bnode) {
TEST_F(MndTestBnode, 03_Drop_Bnode) {
{
// create dnode
int32_t contLen = sizeof(SCreateDnodeMsg);
SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
@ -110,7 +113,6 @@ TEST_F(MndTestBnode, 04_Create_Bnode) {
}
{
// create bnode
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
@ -133,7 +135,6 @@ TEST_F(MndTestBnode, 04_Create_Bnode) {
}
{
// drop bnode
int32_t contLen = sizeof(SMDropBnodeReq);
SMDropBnodeReq* pReq = (SMDropBnodeReq*)rpcMallocCont(contLen);
@ -151,4 +152,143 @@ TEST_F(MndTestBnode, 04_Create_Bnode) {
CheckBinary("localhost:9018", TSDB_EP_LEN);
CheckTimestamp();
}
{
int32_t contLen = sizeof(SMDropBnodeReq);
SMDropBnodeReq* pReq = (SMDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_BNODE_NOT_EXIST);
}
}
TEST_F(MndTestBnode, 03_Create_Bnode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, bnode is creating
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// continue send message, bnode is creating
int32_t contLen = sizeof(SMDropBnodeReq);
SMDropBnodeReq* pReq = (SMDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
if (pMsg->code == 0) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}
TEST_F(MndTestBnode, 04_Drop_Bnode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMDropBnodeReq);
SMDropBnodeReq* pReq = (SMDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, bnode is dropping
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// continue send message, bnode is dropping
int32_t contLen = sizeof(SMDropBnodeReq);
SMDropBnodeReq* pReq = (SMDropBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateBnodeReq);
SMCreateBnodeReq* pReq = (SMCreateBnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_BNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
if (pMsg->code == 0) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}

146
source/dnode/mnode/impl/test/qnode/qnode.cpp
View File

@ -90,13 +90,12 @@ TEST_F(MndTestQnode, 02_Create_Qnode) {
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_DNODE_NOT_EXIST);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_QNODE_ALREADY_EXIST);
}
}
TEST_F(MndTestQnode, 04_Create_Qnode) {
TEST_F(MndTestQnode, 03_Drop_Qnode) {
{
// create dnode
int32_t contLen = sizeof(SCreateDnodeMsg);
SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
@ -114,7 +113,6 @@ TEST_F(MndTestQnode, 04_Create_Qnode) {
}
{
// create qnode
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
@ -137,7 +135,6 @@ TEST_F(MndTestQnode, 04_Create_Qnode) {
}
{
// drop qnode
int32_t contLen = sizeof(SMDropQnodeReq);
SMDropQnodeReq* pReq = (SMDropQnodeReq*)rpcMallocCont(contLen);
@ -155,4 +152,143 @@ TEST_F(MndTestQnode, 04_Create_Qnode) {
CheckBinary("localhost:9014", TSDB_EP_LEN);
CheckTimestamp();
}
{
int32_t contLen = sizeof(SMDropQnodeReq);
SMDropQnodeReq* pReq = (SMDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_QNODE_NOT_EXIST);
}
}
TEST_F(MndTestQnode, 03_Create_Qnode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, qnode is creating
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// continue send message, qnode is creating
int32_t contLen = sizeof(SMDropQnodeReq);
SMDropQnodeReq* pReq = (SMDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
if (pMsg->code == 0) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}
TEST_F(MndTestQnode, 04_Drop_Qnode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMDropQnodeReq);
SMDropQnodeReq* pReq = (SMDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, qnode is dropping
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// continue send message, qnode is dropping
int32_t contLen = sizeof(SMDropQnodeReq);
SMDropQnodeReq* pReq = (SMDropQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateQnodeReq);
SMCreateQnodeReq* pReq = (SMCreateQnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_QNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
if (pMsg->code == 0) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}

168
source/dnode/mnode/impl/test/snode/snode.cpp
View File

@ -50,7 +50,18 @@ TEST_F(MndTestSnode, 01_Show_Snode) {
EXPECT_EQ(test.GetShowRows(), 0);
}
TEST_F(MndTestSnode, 02_Create_Snode_Invalid_Id) {
TEST_F(MndTestSnode, 02_Create_Snode) {
{
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_DNODE_NOT_EXIST);
}
{
int32_t contLen = sizeof(SMCreateSnodeReq);
@ -63,11 +74,6 @@ TEST_F(MndTestSnode, 02_Create_Snode_Invalid_Id) {
test.SendShowMetaMsg(TSDB_MGMT_TABLE_SNODE, "");
CHECK_META("show snodes", 3);
CHECK_SCHEMA(0, TSDB_DATA_TYPE_SMALLINT, 2, "id");
CHECK_SCHEMA(1, TSDB_DATA_TYPE_BINARY, TSDB_EP_LEN + VARSTR_HEADER_SIZE, "endpoint");
CHECK_SCHEMA(2, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
test.SendShowRetrieveMsg();
EXPECT_EQ(test.GetShowRows(), 1);
@ -75,24 +81,21 @@ TEST_F(MndTestSnode, 02_Create_Snode_Invalid_Id) {
CheckBinary("localhost:9016", TSDB_EP_LEN);
CheckTimestamp();
}
}
TEST_F(MndTestSnode, 03_Create_Snode_Invalid_Id) {
{
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
pReq->dnodeId = htonl(1);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_DNODE_NOT_EXIST);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_SNODE_ALREADY_EXIST);
}
}
TEST_F(MndTestSnode, 04_Create_Snode) {
TEST_F(MndTestSnode, 03_Drop_Snode) {
{
// create dnode
int32_t contLen = sizeof(SCreateDnodeMsg);
SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
@ -110,7 +113,6 @@ TEST_F(MndTestSnode, 04_Create_Snode) {
}
{
// create snode
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
@ -133,7 +135,6 @@ TEST_F(MndTestSnode, 04_Create_Snode) {
}
{
// drop snode
int32_t contLen = sizeof(SMDropSnodeReq);
SMDropSnodeReq* pReq = (SMDropSnodeReq*)rpcMallocCont(contLen);
@ -151,4 +152,143 @@ TEST_F(MndTestSnode, 04_Create_Snode) {
CheckBinary("localhost:9016", TSDB_EP_LEN);
CheckTimestamp();
}
{
int32_t contLen = sizeof(SMDropSnodeReq);
SMDropSnodeReq* pReq = (SMDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_MND_SNODE_NOT_EXIST);
}
}
TEST_F(MndTestSnode, 03_Create_Snode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, snode is creating
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// continue send message, snode is creating
int32_t contLen = sizeof(SMDropSnodeReq);
SMDropSnodeReq* pReq = (SMDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_CREATING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
if (pMsg->code == 0) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}
TEST_F(MndTestSnode, 04_Drop_Snode_Rollback) {
{
// send message first, then dnode2 crash, result is returned, and rollback is started
int32_t contLen = sizeof(SMDropSnodeReq);
SMDropSnodeReq* pReq = (SMDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_RPC_NETWORK_UNAVAIL);
}
{
// continue send message, snode is dropping
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// continue send message, snode is dropping
int32_t contLen = sizeof(SMDropSnodeReq);
SMDropSnodeReq* pReq = (SMDropSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
server2.Stop();
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_DROP_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, TSDB_CODE_SDB_OBJ_DROPPING);
}
{
// server start, wait until the rollback finished
server2.DoStart();
taosMsleep(1000);
int32_t retry = 0;
int32_t retryMax = 10;
for (retry = 0; retry < retryMax; retry++) {
int32_t contLen = sizeof(SMCreateSnodeReq);
SMCreateSnodeReq* pReq = (SMCreateSnodeReq*)rpcMallocCont(contLen);
pReq->dnodeId = htonl(2);
SRpcMsg* pMsg = test.SendMsg(TDMT_MND_CREATE_SNODE, pReq, contLen);
ASSERT_NE(pMsg, nullptr);
if (pMsg->code == 0) break;
taosMsleep(1000);
}
ASSERT_NE(retry, retryMax);
}
}

4
source/dnode/mnode/sdb/src/sdbHash.c
View File

@ -69,6 +69,8 @@ static const char *sdbStatusStr(ESdbStatus status) {
return "ready";
case SDB_STATUS_DROPPED:
return "dropped";
case SDB_STATUS_INIT:
return "init";
default:
return "undefine";
}
@ -261,6 +263,8 @@ int32_t sdbWrite(SSdb *pSdb, SSdbRaw *pRaw) {
}
void *sdbAcquire(SSdb *pSdb, ESdbType type, void *pKey) {
terrno = 0;
SHashObj *hash = sdbGetHash(pSdb, type);
if (hash == NULL) return NULL;

8
source/libs/catalog/inc/catalogInt.h
View File

@ -64,6 +64,14 @@ typedef struct SCatalogMgmt {
typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
#define CTG_IS_STABLE(isSTable) (1 == (isSTable))
#define CTG_IS_NOT_STABLE(isSTable) (0 == (isSTable))
#define CTG_IS_UNKNOWN_STABLE(isSTable) ((isSTable) < 0)
#define CTG_SET_STABLE(isSTable, tbType) do { (isSTable) = ((tbType) == TSDB_SUPER_TABLE) ? 1 : ((tbType) > TSDB_SUPER_TABLE ? 0 : -1); } while (0)
#define CTG_TBTYPE_MATCH(isSTable, tbType) (CTG_IS_UNKNOWN_STABLE(isSTable) || (CTG_IS_STABLE(isSTable) && (tbType) == TSDB_SUPER_TABLE) || (CTG_IS_NOT_STABLE(isSTable) && (tbType) != TSDB_SUPER_TABLE))
#define CTG_TABLE_NOT_EXIST(code) (code == TSDB_CODE_TDB_INVALID_TABLE_ID)
#define ctgFatal(...) do { if (ctgDebugFlag & DEBUG_FATAL) { taosPrintLog("CTG FATAL ", ctgDebugFlag, __VA_ARGS__); }} while(0)
#define ctgError(...) do { if (ctgDebugFlag & DEBUG_ERROR) { taosPrintLog("CTG ERROR ", ctgDebugFlag, __VA_ARGS__); }} while(0)
#define ctgWarn(...) do { if (ctgDebugFlag & DEBUG_WARN) { taosPrintLog("CTG WARN ", ctgDebugFlag, __VA_ARGS__); }} while(0)

167
source/libs/catalog/src/catalog.c
View File

@ -105,6 +105,8 @@ int32_t ctgGetTableMetaFromCache(struct SCatalog* pCatalog, const SName* pTableN
}
*exist = 1;
tbMeta = *pTableMeta;
if (tbMeta->tableType != TSDB_CHILD_TABLE) {
return TSDB_CODE_SUCCESS;
@ -143,6 +145,29 @@ int32_t ctgGetTableMetaFromCache(struct SCatalog* pCatalog, const SName* pTableN
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTableTypeFromCache(struct SCatalog* pCatalog, const SName* pTableName, int32_t *tbType) {
if (NULL == pCatalog->tableCache.cache) {
return TSDB_CODE_SUCCESS;
}
char tbFullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pTableName, tbFullName);
size_t sz = 0;
STableMeta *pTableMeta = NULL;
taosHashGetCloneExt(pCatalog->tableCache.cache, tbFullName, strlen(tbFullName), NULL, (void **)&pTableMeta, &sz);
if (NULL == pTableMeta) {
return TSDB_CODE_SUCCESS;
}
*tbType = pTableMeta->tableType;
return TSDB_CODE_SUCCESS;
}
void ctgGenEpSet(SEpSet *epSet, SVgroupInfo *vgroupInfo) {
epSet->inUse = 0;
epSet->numOfEps = vgroupInfo->numOfEps;
@ -153,14 +178,7 @@ void ctgGenEpSet(SEpSet *epSet, SVgroupInfo *vgroupInfo) {
}
}
int32_t ctgGetTableMetaFromMnode(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, STableMetaOutput* output) {
if (NULL == pCatalog || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == output) {
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
char tbFullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pTableName, tbFullName);
int32_t ctgGetTableMetaFromMnodeImpl(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, char* tbFullName, STableMetaOutput* output) {
SBuildTableMetaInput bInput = {.vgId = 0, .dbName = NULL, .tableFullName = tbFullName};
char *msg = NULL;
SEpSet *pVnodeEpSet = NULL;
@ -179,6 +197,12 @@ int32_t ctgGetTableMetaFromMnode(struct SCatalog* pCatalog, void *pRpc, const SE
rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
if (TSDB_CODE_SUCCESS != rpcRsp.code) {
if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
output->metaNum = 0;
ctgDebug("tbmeta:%s not exist in mnode", tbFullName);
return TSDB_CODE_SUCCESS;
}
ctgError("error rsp for table meta, code:%x", rpcRsp.code);
CTG_ERR_RET(rpcRsp.code);
}
@ -188,6 +212,13 @@ int32_t ctgGetTableMetaFromMnode(struct SCatalog* pCatalog, void *pRpc, const SE
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTableMetaFromMnode(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, STableMetaOutput* output) {
char tbFullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pTableName, tbFullName);
return ctgGetTableMetaFromMnodeImpl(pCatalog, pRpc, pMgmtEps, tbFullName, output);
}
int32_t ctgGetTableMetaFromVnode(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* output) {
if (NULL == pCatalog || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == vgroupInfo || NULL == output) {
@ -197,7 +228,7 @@ int32_t ctgGetTableMetaFromVnode(struct SCatalog* pCatalog, void *pRpc, const SE
char dbFullName[TSDB_DB_FNAME_LEN];
tNameGetFullDbName(pTableName, dbFullName);
SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbName = dbFullName, .tableFullName = pTableName->tname};
SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbName = dbFullName, .tableFullName = (char *)pTableName->tname};
char *msg = NULL;
SEpSet *pVnodeEpSet = NULL;
int32_t msgLen = 0;
@ -214,10 +245,15 @@ int32_t ctgGetTableMetaFromVnode(struct SCatalog* pCatalog, void *pRpc, const SE
SEpSet epSet;
ctgGenEpSet(&epSet, vgroupInfo);
rpcSendRecv(pRpc, &epSet, &rpcMsg, &rpcRsp);
if (TSDB_CODE_SUCCESS != rpcRsp.code) {
if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
output->metaNum = 0;
ctgDebug("tbmeta:%s not exist in vnode", pTableName->tname);
return TSDB_CODE_SUCCESS;
}
ctgError("error rsp for table meta, code:%x", rpcRsp.code);
CTG_ERR_RET(rpcRsp.code);
}
@ -322,22 +358,28 @@ _return:
CTG_RET(TSDB_CODE_SUCCESS);
}
int32_t ctgGetTableMetaImpl(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, bool forceUpdate, STableMeta** pTableMeta) {
int32_t ctgGetTableMetaImpl(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, bool forceUpdate, STableMeta** pTableMeta, int32_t isSTable) {
if (NULL == pCatalog || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
int32_t exist = 0;
if (!forceUpdate) {
if (!forceUpdate) {
CTG_ERR_RET(ctgGetTableMetaFromCache(pCatalog, pTableName, pTableMeta, &exist));
if (exist) {
if (exist && CTG_TBTYPE_MATCH(isSTable, (*pTableMeta)->tableType)) {
return TSDB_CODE_SUCCESS;
}
} else if (CTG_IS_UNKNOWN_STABLE(isSTable)) {
int32_t tbType = 0;
CTG_ERR_RET(ctgGetTableTypeFromCache(pCatalog, pTableName, &tbType));
CTG_SET_STABLE(isSTable, tbType);
}
CTG_ERR_RET(catalogRenewTableMeta(pCatalog, pRpc, pMgmtEps, pTableName));
CTG_ERR_RET(ctgRenewTableMetaImpl(pCatalog, pRpc, pMgmtEps, pTableName, isSTable));
CTG_ERR_RET(ctgGetTableMetaFromCache(pCatalog, pTableName, pTableMeta, &exist));
@ -364,19 +406,27 @@ int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *out
}
if (NULL == pCatalog->tableCache.cache) {
pCatalog->tableCache.cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
if (NULL == pCatalog->tableCache.cache) {
SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
if (NULL == cache) {
ctgError("init hash[%d] for tablemeta cache failed", ctgMgmt.cfg.maxTblCacheNum);
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
}
if (NULL != atomic_val_compare_exchange_ptr(&pCatalog->tableCache.cache, NULL, cache)) {
taosHashCleanup(cache);
}
}
if (NULL == pCatalog->tableCache.stableCache) {
pCatalog->tableCache.stableCache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
if (NULL == pCatalog->tableCache.stableCache) {
SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
if (NULL == cache) {
ctgError("init hash[%d] for stablemeta cache failed", ctgMgmt.cfg.maxTblCacheNum);
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
}
if (NULL != atomic_val_compare_exchange_ptr(&pCatalog->tableCache.stableCache, NULL, cache)) {
taosHashCleanup(cache);
}
}
if (output->metaNum == 2) {
@ -481,6 +531,50 @@ int32_t ctgValidateAndRemoveDb(struct SCatalog* pCatalog, const char* dbName, SD
return TSDB_CODE_SUCCESS;
}
int32_t ctgRenewTableMetaImpl(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable) {
if (NULL == pCatalog || NULL == pTransporter || NULL == pMgmtEps || NULL == pTableName) {
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
SVgroupInfo vgroupInfo = {0};
int32_t code = 0;
CTG_ERR_RET(catalogGetTableHashVgroup(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo));
STableMetaOutput voutput = {0};
STableMetaOutput moutput = {0};
STableMetaOutput *output = &voutput;
if (CTG_IS_STABLE(isSTable)) {
CTG_ERR_JRET(ctgGetTableMetaFromMnode(pCatalog, pTransporter, pMgmtEps, pTableName, &moutput));
if (0 == moutput.metaNum) {
CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo, &voutput));
} else {
output = &moutput;
}
} else {
CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo, &voutput));
if (voutput.metaNum > 0 && TSDB_SUPER_TABLE == voutput.tbMeta->tableType) {
CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCatalog, pTransporter, pMgmtEps, voutput.tbFname, &moutput));
tfree(voutput.tbMeta);
voutput.tbMeta = moutput.tbMeta;
moutput.tbMeta = NULL;
}
}
CTG_ERR_JRET(ctgUpdateTableMetaCache(pCatalog, output));
_return:
tfree(voutput.tbMeta);
tfree(moutput.tbMeta);
CTG_RET(code);
}
int32_t catalogInit(SCatalogCfg *cfg) {
if (ctgMgmt.pCluster) {
@ -643,11 +737,15 @@ int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDB
}
if (NULL == pCatalog->dbCache.cache) {
pCatalog->dbCache.cache = taosHashInit(ctgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
if (NULL == pCatalog->dbCache.cache) {
SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
if (NULL == cache) {
ctgError("init hash[%d] for db cache failed", CTG_DEFAULT_CACHE_DB_NUMBER);
CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
}
if (NULL != atomic_val_compare_exchange_ptr(&pCatalog->dbCache.cache, NULL, cache)) {
taosHashCleanup(cache);
}
} else {
CTG_ERR_JRET(ctgValidateAndRemoveDb(pCatalog, dbName, dbInfo));
}
@ -672,34 +770,23 @@ _return:
}
int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
return ctgGetTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, false, pTableMeta);
return ctgGetTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, false, pTableMeta, -1);
}
int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName) {
int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
return ctgGetTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, false, pTableMeta, 1);
}
int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable) {
if (NULL == pCatalog || NULL == pTransporter || NULL == pMgmtEps || NULL == pTableName) {
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
SVgroupInfo vgroupInfo = {0};
int32_t code = 0;
CTG_ERR_RET(catalogGetTableHashVgroup(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo));
STableMetaOutput output = {0};
CTG_ERR_RET(ctgGetTableMetaFromVnode(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo, &output));
//CTG_ERR_RET(ctgGetTableMetaFromMnode(pCatalog, pRpc, pMgmtEps, pTableName, &output));
CTG_ERR_JRET(ctgUpdateTableMetaCache(pCatalog, &output));
_return:
tfree(output.tbMeta);
CTG_RET(code);
return ctgRenewTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, isSTable);
}
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
return ctgGetTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, true, pTableMeta);
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable) {
return ctgGetTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, true, pTableMeta, isSTable);
}
int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList) {

7
source/libs/catalog/test/catalogTests.cpp
View File

@ -45,7 +45,7 @@ void ctgTestSetPrepareSTableMeta();
bool ctgTestStop = false;
bool ctgTestEnableSleep = false;
bool ctgTestDeadLoop = true;
bool ctgTestDeadLoop = false;
int32_t ctgTestCurrentVgVersion = 0;
int32_t ctgTestVgVersion = 1;
@ -600,7 +600,6 @@ void *ctgTestSetCtableMetaThread(void *param) {
}
#if 0
TEST(tableMeta, normalTable) {
struct SCatalog* pCtg = NULL;
@ -768,7 +767,7 @@ TEST(tableMeta, superTableCase) {
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
tableMeta = NULL;
code = catalogRenewAndGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta);
code = catalogRenewAndGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta, 0);
ASSERT_EQ(code, 0);
ASSERT_EQ(tableMeta->vgId, 9);
ASSERT_EQ(tableMeta->tableType, TSDB_CHILD_TABLE);
@ -999,8 +998,6 @@ TEST(multiThread, getSetDbVgroupCase) {
catalogDestroy();
}
#endif
TEST(multiThread, ctableMeta) {
struct SCatalog* pCtg = NULL;

2
source/libs/parser/inc/astToMsg.h
View File

@ -10,7 +10,7 @@ SCreateAcctReq* buildAcctManipulationMsg(SSqlInfo* pInfo, int32_t* outputLen, in
SDropUserReq* buildDropUserMsg(SSqlInfo* pInfo, int32_t* outputLen, int64_t id, char* msgBuf, int32_t msgLen);
SShowMsg* buildShowMsg(SShowInfo* pShowInfo, SParseBasicCtx* pParseCtx, char* msgBuf, int32_t msgLen);
SCreateDbMsg* buildCreateDbMsg(SCreateDbInfo* pCreateDbInfo, SParseBasicCtx *pCtx, SMsgBuf* pMsgBuf);
SCreateStbMsg* buildCreateTableMsg(SCreateTableSql* pCreateTableSql, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf);
SCreateStbMsg* buildCreateStbMsg(SCreateTableSql* pCreateTableSql, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf);
SDropStbMsg* buildDropStableMsg(SSqlInfo* pInfo, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf);
SCreateDnodeMsg *buildCreateDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMsgBuf);
SDropDnodeMsg *buildDropDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMsgBuf);

8
source/libs/parser/inc/parserInt.h
View File

@ -38,14 +38,6 @@ typedef struct SMsgBuf {
char *buf;
} SMsgBuf;
// create table operation type
enum TSQL_CREATE_TABLE_TYPE {
TSQL_CREATE_TABLE = 0x1,
TSQL_CREATE_STABLE = 0x2,
TSQL_CREATE_CTABLE = 0x3,
TSQL_CREATE_STREAM = 0x4,
};
void clearTableMetaInfo(STableMetaInfo* pTableMetaInfo);
void clearAllTableMetaInfo(SQueryStmtInfo* pQueryInfo, bool removeMeta, uint64_t id);

16
source/libs/parser/inc/sql.y
View File

@ -370,7 +370,7 @@ create_table_list(A) ::= create_from_stable(Z). {
pCreateTable->childTableInfo = taosArrayInit(4, sizeof(SCreatedTableInfo));
taosArrayPush(pCreateTable->childTableInfo, &Z);
pCreateTable->type = TSQL_CREATE_CTABLE;
pCreateTable->type = TSDB_SQL_CREATE_TABLE;
A = pCreateTable;
}
@ -381,7 +381,7 @@ create_table_list(A) ::= create_table_list(X) create_from_stable(Z). {
%type create_table_args{SCreateTableSql*}
create_table_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) LP columnlist(X) RP. {
A = tSetCreateTableInfo(X, NULL, NULL, TSQL_CREATE_TABLE);
A = tSetCreateTableInfo(X, NULL, NULL, TSDB_SQL_CREATE_TABLE);
setSqlInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE);
V.n += Z.n;
@ -391,7 +391,7 @@ create_table_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) LP columnlist(X) RP. {
// create super table
%type create_stable_args{SCreateTableSql*}
create_stable_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) LP columnlist(X) RP TAGS LP columnlist(Y) RP. {
A = tSetCreateTableInfo(X, Y, NULL, TSQL_CREATE_STABLE);
A = tSetCreateTableInfo(X, Y, NULL, TSDB_SQL_CREATE_STABLE);
setSqlInfo(pInfo, A, NULL, TSDB_SQL_CREATE_STABLE);
V.n += Z.n;
@ -421,11 +421,11 @@ tagNamelist(A) ::= ids(X). {A = taosArrayInit(4, sizeof(STo
// create stream
// create table table_name as select count(*) from super_table_name interval(time)
create_table_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) AS select(S). {
A = tSetCreateTableInfo(NULL, NULL, S, TSQL_CREATE_STREAM);
setSqlInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE);
V.n += Z.n;
setCreatedTableName(pInfo, &V, &U);
// A = tSetCreateTableInfo(NULL, NULL, S, TSQL_CREATE_STREAM);
// setSqlInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE);
//
// V.n += Z.n;
// setCreatedTableName(pInfo, &V, &U);
}
%type column{SField}

18
source/libs/parser/src/astGenerator.c
View File

@ -579,25 +579,21 @@ SCreateTableSql *tSetCreateTableInfo(SArray *pCols, SArray *pTags, SSqlNode *pSe
SCreateTableSql *pCreate = calloc(1, sizeof(SCreateTableSql));
switch (type) {
case TSQL_CREATE_TABLE: {
case TSDB_SQL_CREATE_TABLE: {
pCreate->colInfo.pColumns = pCols;
assert(pTags == NULL);
break;
}
case TSQL_CREATE_STABLE: {
case TSDB_SQL_CREATE_STABLE: {
pCreate->colInfo.pColumns = pCols;
pCreate->colInfo.pTagColumns = pTags;
assert(pTags != NULL && pCols != NULL);
break;
}
case TSQL_CREATE_STREAM: {
pCreate->pSelect = pSelect;
break;
}
case TSQL_CREATE_CTABLE: {
assert(0);
}
// case TSQL_CREATE_STREAM: {
// pCreate->pSelect = pSelect;
// break;
// }
default:
assert(false);
@ -785,7 +781,7 @@ void destroySqlInfo(SSqlInfo *pInfo) {
taosArrayDestroy(pInfo->funcs);
if (pInfo->type == TSDB_SQL_SELECT) {
destroyAllSqlNode(&pInfo->sub);
} else if (pInfo->type == TSDB_SQL_CREATE_STABLE) {
} else if (pInfo->type == TSDB_SQL_CREATE_STABLE || pInfo->type == TSDB_SQL_CREATE_TABLE) {
pInfo->pCreateTableInfo = destroyCreateTableSql(pInfo->pCreateTableInfo);
} else if (pInfo->type == TSDB_SQL_ALTER_TABLE) {
taosArrayDestroyEx(pInfo->pAlterInfo->varList, freeItem);

29
source/libs/parser/src/astToMsg.c
View File

@ -230,7 +230,7 @@ SCreateDbMsg* buildCreateDbMsg(SCreateDbInfo* pCreateDbInfo, SParseBasicCtx *pCt
return pCreateMsg;
}
SCreateStbMsg* buildCreateTableMsg(SCreateTableSql* pCreateTableSql, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf) {
SCreateStbMsg* buildCreateStbMsg(SCreateTableSql* pCreateTableSql, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf) {
SSchema* pSchema;
int32_t numOfTags = 0;
@ -239,16 +239,16 @@ SCreateStbMsg* buildCreateTableMsg(SCreateTableSql* pCreateTableSql, int32_t* le
numOfTags = (int32_t) taosArrayGetSize(pCreateTableSql->colInfo.pTagColumns);
}
SCreateStbMsg* pCreateTableMsg = (SCreateStbMsg*)calloc(1, sizeof(SCreateStbMsg) + (numOfCols + numOfTags) * sizeof(SSchema));
SCreateStbMsg* pCreateStbMsg = (SCreateStbMsg*)calloc(1, sizeof(SCreateStbMsg) + (numOfCols + numOfTags) * sizeof(SSchema));
char* pMsg = NULL;
#if 0
int32_t tableType = pCreateTableSql->type;
if (tableType != TSQL_CREATE_TABLE && tableType != TSQL_CREATE_STABLE) { // create by using super table, tags value
#if 0
SArray* list = pInfo->pCreateTableInfo->childTableInfo;
int32_t numOfTables = (int32_t)taosArrayGetSize(list);
pCreateTableMsg->numOfTables = htonl(numOfTables);
pCreateStbMsg->numOfTables = htonl(numOfTables);
pMsg = (char*)pCreateMsg;
for (int32_t i = 0; i < numOfTables; ++i) {
@ -268,25 +268,27 @@ SCreateStbMsg* buildCreateTableMsg(SCreateTableSql* pCreateTableSql, int32_t* le
int32_t len = (int32_t)(pMsg - (char*)pCreate);
pCreate->len = htonl(len);
}
} else {
#endif
} else { // create (super) table
// create (super) table
SName n = {0};
int32_t code = createSName(&n, &pCreateTableSql->name, pParseCtx, pMsgBuf);
if (code != 0) {
return NULL;
}
code = tNameExtractFullName(&n, pCreateTableMsg->name);
code = tNameExtractFullName(&n, pCreateStbMsg->name);
if (code != 0) {
buildInvalidOperationMsg(pMsgBuf, "invalid table name or database not specified");
return NULL;
}
pCreateTableMsg->igExists = pCreateTableSql->existCheck ? 1 : 0;
pCreateTableMsg->numOfColumns = htonl(numOfCols);
pCreateTableMsg->numOfTags = htonl(numOfTags);
pCreateStbMsg->igExists = pCreateTableSql->existCheck ? 1 : 0;
pCreateStbMsg->numOfColumns = htonl(numOfCols);
pCreateStbMsg->numOfTags = htonl(numOfTags);
pSchema = (SSchema*) pCreateTableMsg->pSchema;
pSchema = (SSchema*)pCreateStbMsg->pSchema;
for (int i = 0; i < numOfCols; ++i) {
SField* pField = taosArrayGet(pCreateTableSql->colInfo.pColumns, i);
pSchema->type = pField->type;
@ -306,12 +308,11 @@ SCreateStbMsg* buildCreateTableMsg(SCreateTableSql* pCreateTableSql, int32_t* le
}
pMsg = (char*)pSchema;
}
int32_t msgLen = (int32_t)(pMsg - (char*)pCreateTableMsg);
int32_t msgLen = (int32_t)(pMsg - (char*)pCreateStbMsg);
*len = msgLen;
return pCreateTableMsg;
return pCreateStbMsg;
}
SDropStbMsg* buildDropStableMsg(SSqlInfo* pInfo, int32_t* len, SParseBasicCtx* pParseCtx, SMsgBuf* pMsgBuf) {
@ -330,7 +331,7 @@ SDropStbMsg* buildDropStableMsg(SSqlInfo* pInfo, int32_t* len, SParseBasicCtx* p
assert(code == TSDB_CODE_SUCCESS && name.type == TSDB_TABLE_NAME_T);
pDropTableMsg->igNotExists = pInfo->pMiscInfo->existsCheck ? 1 : 0;
*len = sizeof(SDropTableMsg);
*len = sizeof(SDropStbMsg);
return pDropTableMsg;
}

275
source/libs/parser/src/dCDAstProcess.c
View File

@ -1,10 +1,12 @@
#include <astGenerator.h>
#include <tmsg.h>
#include <ttime.h>
#include "astToMsg.h"
#include "parserInt.h"
#include "parserUtil.h"
#include "queryInfoUtil.h"
#include "tglobal.h"
#include "tmsg.h"
#include "ttime.h"
/* is contained in pFieldList or not */
static bool has(SArray* pFieldList, int32_t startIndex, const char* name) {
@ -248,16 +250,10 @@ static int32_t validateTableColumns(SArray* pFieldList, int32_t maxRowLength, in
}
static int32_t validateTableColumnInfo(SArray* pFieldList, SMsgBuf* pMsgBuf) {
assert(pFieldList != NULL);
assert(pFieldList != NULL && pMsgBuf != NULL);
const char* msg1 = "first column must be timestamp";
const char* msg2 = "row length exceeds max length";
const char* msg3 = "duplicated column names";
const char* msg4 = "invalid data type";
const char* msg5 = "invalid binary/nchar column length";
const char* msg6 = "invalid column name";
const char* msg7 = "too many columns";
const char* msg8 = "illegal number of columns";
const char* msg2 = "illegal number of columns";
// first column must be timestamp
SField* pField = taosArrayGet(pFieldList, 0);
@ -268,7 +264,7 @@ static int32_t validateTableColumnInfo(SArray* pFieldList, SMsgBuf* pMsgBuf) {
// number of fields no less than 2
size_t numOfCols = taosArrayGetSize(pFieldList);
if (numOfCols <= 1) {
return buildInvalidOperationMsg(pMsgBuf, msg8);
return buildInvalidOperationMsg(pMsgBuf, msg2);
}
return validateTableColumns(pFieldList, TSDB_MAX_BYTES_PER_ROW, TSDB_MAX_COLUMNS, pMsgBuf);
@ -297,11 +293,9 @@ static int32_t validateTagParams(SArray* pTagsList, SArray* pFieldList, SMsgBuf*
return validateTableColumns(pFieldList, TSDB_MAX_TAGS_LEN, TSDB_MAX_TAGS, pMsgBuf);
}
int32_t doCheckForCreateTable(SSqlInfo* pInfo, SMsgBuf* pMsgBuf) {
int32_t doCheckForCreateTable(SCreateTableSql* pCreateTable, SMsgBuf* pMsgBuf) {
const char* msg1 = "invalid table name";
SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
SArray* pFieldList = pCreateTable->colInfo.pColumns;
SArray* pTagList = pCreateTable->colInfo.pTagColumns;
assert(pFieldList != NULL);
@ -326,6 +320,8 @@ typedef struct SVgroupTablesBatch {
SVgroupInfo info;
} SVgroupTablesBatch;
static SArray* doSerializeVgroupCreateTableInfo(SHashObj* pVgroupHashmap);
static int32_t doParseSerializeTagValue(SSchema* pTagSchema, int32_t numOfInputTag, SKVRowBuilder* pKvRowBuilder,
SArray* pTagValList, int32_t tsPrecision, SMsgBuf* pMsgBuf) {
const char* msg1 = "illegal value or data overflow";
@ -351,14 +347,52 @@ static int32_t doParseSerializeTagValue(SSchema* pTagSchema, int32_t numOfInputT
return code;
}
int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* pMsgBuf, char** pOutput, int32_t* len) {
static void addCreateTbReqIntoVgroup(SHashObj* pVgroupHashmap, const SName* pTableName, SKVRow row, uint64_t suid, SVgroupInfo* pVgInfo) {
struct SVCreateTbReq req = {0};
req.type = TD_CHILD_TABLE;
req.name = strdup(tNameGetTableName(pTableName));
req.ctbCfg.suid = suid;
req.ctbCfg.pTag = row;
SVgroupTablesBatch* pTableBatch = taosHashGet(pVgroupHashmap, &pVgInfo->vgId, sizeof(pVgInfo->vgId));
if (pTableBatch == NULL) {
SVgroupTablesBatch tBatch = {0};
tBatch.info = *pVgInfo;
tBatch.req.pArray = taosArrayInit(4, sizeof(struct SVCreateTbReq));
taosArrayPush(tBatch.req.pArray, &req);
taosHashPut(pVgroupHashmap, &pVgInfo->vgId, sizeof(pVgInfo->vgId), &tBatch, sizeof(tBatch));
} else { // add to the correct vgroup
assert(pVgInfo->vgId == pTableBatch->info.vgId);
taosArrayPush(pTableBatch->req.pArray, &req);
}
}
static void destroyCreateTbReqBatch(SVgroupTablesBatch* pTbBatch) {
size_t size = taosArrayGetSize(pTbBatch->req.pArray);
for(int32_t i = 0; i < size; ++i) {
SVCreateTbReq* pTableReq = taosArrayGet(pTbBatch->req.pArray, i);
tfree(pTableReq->name);
if (pTableReq->type == TSDB_NORMAL_TABLE) {
tfree(pTableReq->ntbCfg.pSchema);
} else if (pTableReq->type == TSDB_CHILD_TABLE) {
tfree(pTableReq->ctbCfg.pTag);
} else {
assert(0);
}
}
taosArrayDestroy(pTbBatch->req.pArray);
}
static int32_t doCheckAndBuildCreateCTableReq(SCreateTableSql* pCreateTable, SParseBasicCtx* pCtx, SMsgBuf* pMsgBuf, SArray** pBufArray) {
const char* msg1 = "invalid table name";
const char* msg2 = "tags number not matched";
const char* msg3 = "tag value too long";
const char* msg4 = "illegal value or data overflow";
SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
SHashObj* pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
// super table name, create table by using dst
@ -378,6 +412,11 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
return code;
}
SKVRowBuilder kvRowBuilder = {0};
if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
SArray* pValList = pCreateTableInfo->pTagVals;
size_t numOfInputTag = taosArrayGetSize(pValList);
@ -393,26 +432,22 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
SSchema* pTagSchema = getTableTagSchema(pSuperTableMeta);
STableComInfo tinfo = getTableInfo(pSuperTableMeta);
SKVRowBuilder kvRowBuilder = {0};
if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
SArray* pNameList = NULL;
size_t nameSize = 0;
size_t numOfBoundTags = 0;
int32_t schemaSize = getNumOfTags(pSuperTableMeta);
if (pCreateTableInfo->pTagNames) {
pNameList = pCreateTableInfo->pTagNames;
nameSize = taosArrayGetSize(pNameList);
numOfBoundTags = taosArrayGetSize(pNameList);
if (numOfInputTag != nameSize || schemaSize < numOfInputTag) {
if (numOfInputTag != numOfBoundTags || schemaSize < numOfInputTag) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tfree(pSuperTableMeta);
return buildInvalidOperationMsg(pMsgBuf, msg2);
}
bool findColumnIndex = false;
for (int32_t i = 0; i < nameSize; ++i) {
for (int32_t i = 0; i < numOfBoundTags; ++i) {
SToken* sToken = taosArrayGet(pNameList, i);
char tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0}; // create tmp buf to avoid alter orginal sqlstr
@ -440,6 +475,7 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
if (pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR) {
if (pItem->pVar.nLen > pSchema->bytes) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tfree(pSuperTableMeta);
return buildInvalidOperationMsg(pMsgBuf, msg3);
}
} else if (pSchema->type == TSDB_DATA_TYPE_TIMESTAMP) {
@ -460,12 +496,14 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
int16_t len = varDataTLen(tagVal);
if (len > pSchema->bytes) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tfree(pSuperTableMeta);
return buildInvalidOperationMsg(pMsgBuf, msg3);
}
}
if (code != TSDB_CODE_SUCCESS) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tfree(pSuperTableMeta);
return buildInvalidOperationMsg(pMsgBuf, msg4);
}
@ -484,11 +522,14 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
} else {
if (schemaSize != numOfInputTag) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tfree(pSuperTableMeta);
return buildInvalidOperationMsg(pMsgBuf, msg2);
}
code = doParseSerializeTagValue(pTagSchema, numOfInputTag, &kvRowBuilder, pValList, tinfo.precision, pMsgBuf);
if (code != TSDB_CODE_SUCCESS) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tfree(pSuperTableMeta);
return code;
}
}
@ -496,6 +537,7 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
if (row == NULL) {
tfree(pSuperTableMeta);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
@ -503,6 +545,84 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
SName tableName = {0};
code = createSName(&tableName, &pCreateTableInfo->name, pCtx, pMsgBuf);
if (code != TSDB_CODE_SUCCESS) {
tfree(pSuperTableMeta);
return code;
}
// Find a appropriate vgroup to accommodate this table , according to the table name
SVgroupInfo info = {0};
catalogGetTableHashVgroup(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, &tableName, &info);
addCreateTbReqIntoVgroup(pVgroupHashmap, &tableName, row, pSuperTableMeta->uid, &info);
tfree(pSuperTableMeta);
}
*pBufArray = doSerializeVgroupCreateTableInfo(pVgroupHashmap);
taosHashCleanup(pVgroupHashmap);
return TSDB_CODE_SUCCESS;
}
static int32_t serializeVgroupTablesBatchImpl(SVgroupTablesBatch* pTbBatch, SArray* pBufArray) {
int tlen = sizeof(SMsgHead) + tSVCreateTbBatchReqSerialize(NULL, &(pTbBatch->req));
void* buf = malloc(tlen);
if (buf == NULL) {
// TODO: handle error
}
((SMsgHead*)buf)->vgId = htonl(pTbBatch->info.vgId);
((SMsgHead*)buf)->contLen = htonl(tlen);
void* pBuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
tSVCreateTbBatchReqSerialize(&pBuf, &(pTbBatch->req));
SVgDataBlocks* pVgData = calloc(1, sizeof(SVgDataBlocks));
pVgData->vg = pTbBatch->info;
pVgData->pData = buf;
pVgData->size = tlen;
pVgData->numOfTables = (int32_t) taosArrayGetSize(pTbBatch->req.pArray);
taosArrayPush(pBufArray, &pVgData);
}
static int32_t doBuildSingleTableBatchReq(SName* pTableName, SArray* pColumns, SVgroupInfo* pVgroupInfo, SVgroupTablesBatch* pBatch) {
struct SVCreateTbReq req = {0};
req.type = TD_NORMAL_TABLE;
req.name = strdup(tNameGetTableName(pTableName));
req.ntbCfg.nCols = taosArrayGetSize(pColumns);
int32_t num = req.ntbCfg.nCols;
req.ntbCfg.pSchema = calloc(num, sizeof(SSchema));
for(int32_t i = 0; i < num; ++i) {
SSchema* pSchema = taosArrayGet(pColumns, i);
memcpy(&req.ntbCfg.pSchema[i], pSchema, sizeof(SSchema));
}
pBatch->info = *pVgroupInfo;
pBatch->req.pArray = taosArrayInit(1, sizeof(struct SVCreateTbReq));
if (pBatch->req.pArray == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
taosArrayPush(pBatch->req.pArray, &req);
return TSDB_CODE_SUCCESS;
}
int32_t doCheckAndBuildCreateTableReq(SCreateTableSql* pCreateTable, SParseBasicCtx* pCtx, SMsgBuf* pMsgBuf, char** pOutput, int32_t* len) {
SArray* pBufArray = NULL;
// it is a sql statement to create a normal table
if (pCreateTable->childTableInfo == NULL) {
assert(taosArrayGetSize(pCreateTable->colInfo.pColumns) > 0 && pCreateTable->colInfo.pTagColumns == NULL);
int32_t code = doCheckForCreateTable(pCreateTable, pMsgBuf);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
SName tableName = {0};
code = createSName(&tableName, &pCreateTable->name, pCtx, pMsgBuf);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
@ -510,64 +630,49 @@ int32_t doCheckForCreateCTable(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SMsgBuf* p
SVgroupInfo info = {0};
catalogGetTableHashVgroup(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, &tableName, &info);
struct SVCreateTbReq req = {0};
req.type = TD_CHILD_TABLE;
req.name = strdup(tNameGetTableName(&tableName));
req.ctbCfg.suid = pSuperTableMeta->uid;
req.ctbCfg.pTag = row;
SVgroupTablesBatch* pTableBatch = taosHashGet(pVgroupHashmap, &info.vgId, sizeof(info.vgId));
if (pTableBatch == NULL) {
SVgroupTablesBatch tBatch = {0};
tBatch.info = info;
tBatch.req.pArray = taosArrayInit(4, sizeof(struct SVCreateTbReq));
taosArrayPush(tBatch.req.pArray, &req);
taosHashPut(pVgroupHashmap, &info.vgId, sizeof(info.vgId), &tBatch, sizeof(tBatch));
} else { // add to the correct vgroup
assert(info.vgId == pTableBatch->info.vgId);
taosArrayPush(pTableBatch->req.pArray, &req);
SVgroupTablesBatch tbatch = {0};
code = doBuildSingleTableBatchReq(&tableName, pCreateTable->colInfo.pColumns, &info, &tbatch);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
pBufArray = taosArrayInit(1, POINTER_BYTES);
if (pBufArray == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
serializeVgroupTablesBatchImpl(&tbatch, pBufArray);
destroyCreateTbReqBatch(&tbatch);
} else { // it is a child table, created according to a super table
doCheckAndBuildCreateCTableReq(pCreateTable, pCtx, pMsgBuf, &pBufArray);
}
// TODO: serialize and
SVnodeModifOpStmtInfo* pStmtInfo = calloc(1, sizeof(SVnodeModifOpStmtInfo));
pStmtInfo->nodeType = TSDB_SQL_CREATE_TABLE;
pStmtInfo->pDataBlocks = pBufArray;
*pOutput = (char*) pStmtInfo;
*len = sizeof(SVnodeModifOpStmtInfo);
return TSDB_CODE_SUCCESS;
}
SArray* doSerializeVgroupCreateTableInfo(SHashObj* pVgroupHashmap) {
SArray* pBufArray = taosArrayInit(taosHashGetSize(pVgroupHashmap), sizeof(void*));
SVgroupTablesBatch* pTbBatch = NULL;
do {
pTbBatch = taosHashIterate(pVgroupHashmap, pTbBatch);
if (pTbBatch == NULL) break;
int tlen = sizeof(SMsgHead) + tSVCreateTbBatchReqSerialize(NULL, &(pTbBatch->req));
void* buf = malloc(tlen);
if (buf == NULL) {
// TODO: handle error
if (pTbBatch == NULL) {
break;
}
((SMsgHead*)buf)->vgId = htonl(pTbBatch->info.vgId);
((SMsgHead*)buf)->contLen = htonl(tlen);
void* pBuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
tSVCreateTbBatchReqSerialize(&pBuf, &(pTbBatch->req));
SVgDataBlocks* pVgData = calloc(1, sizeof(SVgDataBlocks));
pVgData->vg = pTbBatch->info;
pVgData->pData = buf;
pVgData->size = tlen;
pVgData->numOfTables = (int32_t) taosArrayGetSize(pTbBatch->req.pArray);
taosArrayPush(pBufArray, &pVgData);
/*int32_t code = */serializeVgroupTablesBatchImpl(pTbBatch, pBufArray);
destroyCreateTbReqBatch(pTbBatch);
} while (true);
SVnodeModifOpStmtInfo* pStmtInfo = calloc(1, sizeof(SVnodeModifOpStmtInfo));
pStmtInfo->nodeType = TSDB_SQL_CREATE_TABLE;
pStmtInfo->pDataBlocks = pBufArray;
*pOutput = (char*) pStmtInfo;
*len = sizeof(SVnodeModifOpStmtInfo);
return TSDB_CODE_SUCCESS;
return pBufArray;
}
SDclStmtInfo* qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, char* msgBuf, int32_t msgBufLen) {
@ -776,21 +881,13 @@ SDclStmtInfo* qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, c
case TSDB_SQL_CREATE_STABLE: {
SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
assert(pCreateTable->type != TSQL_CREATE_CTABLE);
if (pCreateTable->type == TSQL_CREATE_TABLE || pCreateTable->type == TSQL_CREATE_STABLE) {
if ((code = doCheckForCreateTable(pInfo, pMsgBuf)) != TSDB_CODE_SUCCESS) {
terrno = code;
goto _error;
}
pDcl->pMsg = (char*)buildCreateTableMsg(pCreateTable, &pDcl->msgLen, pCtx, pMsgBuf);
pDcl->msgType = (pCreateTable->type == TSQL_CREATE_TABLE) ? TDMT_VND_CREATE_TABLE : TDMT_MND_CREATE_STB;
} else if (pCreateTable->type == TSQL_CREATE_STREAM) {
// if ((code = doCheckForStream(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
// return code;
if ((code = doCheckForCreateTable(pCreateTable, pMsgBuf)) != TSDB_CODE_SUCCESS) {
terrno = code;
goto _error;
}
pDcl->pMsg = (char*)buildCreateStbMsg(pCreateTable, &pDcl->msgLen, pCtx, pMsgBuf);
pDcl->msgType = TDMT_MND_CREATE_STB;
break;
}
@ -838,19 +935,19 @@ SDclStmtInfo* qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, c
SVnodeModifOpStmtInfo* qParserValidateCreateTbSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, char* msgBuf, int32_t msgBufLen) {
SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
assert(pCreateTable->type == TSQL_CREATE_CTABLE);
assert(pCreateTable->type == TSDB_SQL_CREATE_TABLE);
SMsgBuf m = {.buf = msgBuf, .len = msgBufLen};
SMsgBuf* pMsgBuf = &m;
SVnodeModifOpStmtInfo* pInsertStmt = NULL;
SVnodeModifOpStmtInfo* pModifSqlStmt = NULL;
int32_t msgLen = 0;
int32_t code = doCheckForCreateCTable(pInfo, pCtx, pMsgBuf, (char**) &pInsertStmt, &msgLen);
int32_t code = doCheckAndBuildCreateTableReq(pCreateTable, pCtx, pMsgBuf, (char**) &pModifSqlStmt, &msgLen);
if (code != TSDB_CODE_SUCCESS) {
tfree(pInsertStmt);
tfree(pModifSqlStmt);
return NULL;
}
return pInsertStmt;
return pModifSqlStmt;
}

28
source/libs/parser/src/parser.c
View File

@ -44,21 +44,21 @@ int32_t parseQuerySql(SParseContext* pCxt, SQueryNode** pQuery) {
}
if (!isDqlSqlStatement(&info)) {
bool toVnode = false;
// bool toVnode = false;
if (info.type == TSDB_SQL_CREATE_TABLE) {
SCreateTableSql* pCreateSql = info.pCreateTableInfo;
if (pCreateSql->type == TSQL_CREATE_CTABLE || pCreateSql->type == TSQL_CREATE_TABLE) {
toVnode = true;
}
}
// SCreateTableSql* pCreateSql = info.pCreateTableInfo;
// if (pCreateSql->type == TSQL_CREATE_CTABLE || pCreateSql->type == TSQL_CREATE_TABLE) {
// toVnode = true;
// }
// }
if (toVnode) {
SVnodeModifOpStmtInfo *pInsertInfo = qParserValidateCreateTbSqlNode(&info, &pCxt->ctx, pCxt->pMsg, pCxt->msgLen);
if (pInsertInfo == NULL) {
// if (toVnode) {
SVnodeModifOpStmtInfo * pModifStmtInfo = qParserValidateCreateTbSqlNode(&info, &pCxt->ctx, pCxt->pMsg, pCxt->msgLen);
if (pModifStmtInfo == NULL) {
return terrno;
}
*pQuery = (SQueryNode*) pInsertInfo;
*pQuery = (SQueryNode*)pModifStmtInfo;
} else {
SDclStmtInfo* pDcl = qParserValidateDclSqlNode(&info, &pCxt->ctx, pCxt->pMsg, pCxt->msgLen);
if (pDcl == NULL) {
@ -240,6 +240,10 @@ void qParserCleanupMetaRequestInfo(SCatalogReq* pMetaReq) {
taosArrayDestroy(pMetaReq->pUdf);
}
void qDestroyQuery(SQueryNode* pQuery) {
// todo
void qDestroyQuery(SQueryNode* pQueryNode) {
if (nodeType(pQueryNode) == TSDB_SQL_INSERT || nodeType(pQueryNode) == TSDB_SQL_CREATE_TABLE) {
SVnodeModifOpStmtInfo* pModifInfo = (SVnodeModifOpStmtInfo*)pQueryNode;
taosArrayDestroy(pModifInfo->pDataBlocks);
}
tfree(pQueryNode);
}

16
source/libs/parser/src/sql.c
View File

@ -2642,7 +2642,7 @@ static void yy_reduce(
pCreateTable->childTableInfo = taosArrayInit(4, sizeof(SCreatedTableInfo));
taosArrayPush(pCreateTable->childTableInfo, &yymsp[0].minor.yy150);
pCreateTable->type = TSQL_CREATE_CTABLE;
pCreateTable->type = TSDB_SQL_CREATE_TABLE;
yylhsminor.yy326 = pCreateTable;
}
yymsp[0].minor.yy326 = yylhsminor.yy326;
@ -2656,7 +2656,7 @@ static void yy_reduce(
break;
case 140: /* create_table_args ::= ifnotexists ids cpxName LP columnlist RP */
{
yylhsminor.yy326 = tSetCreateTableInfo(yymsp[-1].minor.yy165, NULL, NULL, TSQL_CREATE_TABLE);
yylhsminor.yy326 = tSetCreateTableInfo(yymsp[-1].minor.yy165, NULL, NULL, TSDB_SQL_CREATE_TABLE);
setSqlInfo(pInfo, yylhsminor.yy326, NULL, TSDB_SQL_CREATE_TABLE);
yymsp[-4].minor.yy0.n += yymsp[-3].minor.yy0.n;
@ -2666,7 +2666,7 @@ static void yy_reduce(
break;
case 141: /* create_stable_args ::= ifnotexists ids cpxName LP columnlist RP TAGS LP columnlist RP */
{
yylhsminor.yy326 = tSetCreateTableInfo(yymsp[-5].minor.yy165, yymsp[-1].minor.yy165, NULL, TSQL_CREATE_STABLE);
yylhsminor.yy326 = tSetCreateTableInfo(yymsp[-5].minor.yy165, yymsp[-1].minor.yy165, NULL, TSDB_SQL_CREATE_STABLE);
setSqlInfo(pInfo, yylhsminor.yy326, NULL, TSDB_SQL_CREATE_STABLE);
yymsp[-8].minor.yy0.n += yymsp[-7].minor.yy0.n;
@ -2700,11 +2700,11 @@ static void yy_reduce(
break;
case 146: /* create_table_args ::= ifnotexists ids cpxName AS select */
{
yylhsminor.yy326 = tSetCreateTableInfo(NULL, NULL, yymsp[0].minor.yy278, TSQL_CREATE_STREAM);
setSqlInfo(pInfo, yylhsminor.yy326, NULL, TSDB_SQL_CREATE_TABLE);
yymsp[-3].minor.yy0.n += yymsp[-2].minor.yy0.n;
setCreatedTableName(pInfo, &yymsp[-3].minor.yy0, &yymsp[-4].minor.yy0);
// yylhsminor.yy326 = tSetCreateTableInfo(NULL, NULL, yymsp[0].minor.yy278, TSQL_CREATE_STREAM);
// setSqlInfo(pInfo, yylhsminor.yy326, NULL, TSDB_SQL_CREATE_TABLE);
//
// yymsp[-3].minor.yy0.n += yymsp[-2].minor.yy0.n;
// setCreatedTableName(pInfo, &yymsp[-3].minor.yy0, &yymsp[-4].minor.yy0);
}
yymsp[-4].minor.yy326 = yylhsminor.yy326;
break;

12
source/libs/planner/src/logicPlan.c
View File

@ -392,6 +392,18 @@ SArray* createQueryPlanImpl(const SQueryStmtInfo* pQueryInfo) {
}
static void doDestroyQueryNode(SQueryPlanNode* pQueryNode) {
if (pQueryNode->info.type == QNODE_MODIFY) {
SDataPayloadInfo* pInfo = pQueryNode->pExtInfo;
size_t size = taosArrayGetSize(pInfo->payload);
for (int32_t i = 0; i < size; ++i) {
SVgDataBlocks* pBlock = taosArrayGetP(pInfo->payload, i);
tfree(pBlock);
}
taosArrayDestroy(pInfo->payload);
}
tfree(pQueryNode->pExtInfo);
tfree(pQueryNode->pSchema);
tfree(pQueryNode->info.name);

8
source/libs/planner/src/physicalPlan.c
View File

@ -305,10 +305,10 @@ static void splitModificationOpSubPlan(SPlanContext* pCxt, SQueryPlanNode* pPlan
SVgDataBlocks* blocks = (SVgDataBlocks*)taosArrayGetP(pPayload->payload, i);
vgroupInfoToEpSet(&blocks->vg, &subplan->execNode);
subplan->pDataSink = createDataInserter(pCxt, blocks);
subplan->pNode = NULL;
subplan->type = QUERY_TYPE_MODIFY;
subplan->msgType = pPayload->msgType;
subplan->pDataSink = createDataInserter(pCxt, blocks);
subplan->pNode = NULL;
subplan->type = QUERY_TYPE_MODIFY;
subplan->msgType = pPayload->msgType;
subplan->id.queryId = pCxt->pDag->queryId;
RECOVERY_CURRENT_SUBPLAN(pCxt);

36
source/libs/planner/src/planner.c
View File

@ -16,12 +16,44 @@
#include "parser.h"
#include "plannerInt.h"
static void destroyDataSinkNode(SDataSink* pSinkNode) {
if (pSinkNode == NULL) {
return;
}
tfree(pSinkNode);
}
void qDestroySubplan(SSubplan* pSubplan) {
// todo
if (pSubplan == NULL) {
return;
}
taosArrayDestroy(pSubplan->pChildren);
taosArrayDestroy(pSubplan->pParents);
destroyDataSinkNode(pSubplan->pDataSink);
// todo destroy pNode
tfree(pSubplan);
}
void qDestroyQueryDag(struct SQueryDag* pDag) {
// todo
if (pDag == NULL) {
return;
}
size_t size = taosArrayGetSize(pDag->pSubplans);
for(size_t i = 0; i < size; ++i) {
SArray* pa = taosArrayGetP(pDag->pSubplans, i);
size_t t = taosArrayGetSize(pa);
for(int32_t j = 0; j < t; ++j) {
SSubplan* pSubplan = taosArrayGetP(pa, j);
qDestroySubplan(pSubplan);
}
taosArrayDestroy(pa);
}
taosArrayDestroy(pDag->pSubplans);
tfree(pDag);
}
int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, uint64_t requestId) {

5
source/libs/scheduler/inc/schedulerInt.h
View File

@ -114,8 +114,9 @@ typedef struct SSchJob {
#define SCH_IS_DATA_SRC_TASK(task) ((task)->plan->type == QUERY_TYPE_SCAN)
#define SCH_TASK_NEED_WAIT_ALL(task) ((task)->plan->type == QUERY_TYPE_MODIFY)
#define SCH_JOB_ERR_LOG(param, ...) qError("QID:%"PRIx64 param, job->queryId, __VA_ARGS__)
#define SCH_TASK_ERR_LOG(param, ...) qError("QID:%"PRIx64",TID:%"PRIx64 param, job->queryId, task->taskId, __VA_ARGS__)
#define SCH_JOB_ELOG(param, ...) qError("QID:% "PRIx64 param, job->queryId, __VA_ARGS__)
#define SCH_TASK_ELOG(param, ...) qError("QID:%"PRIx64",TID:% "PRIx64 param, job->queryId, task->taskId, __VA_ARGS__)
#define SCH_TASK_DLOG(param, ...) qDebug("QID:%"PRIx64",TID:% "PRIx64 param, job->queryId, task->taskId, __VA_ARGS__)
#define SCH_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define SCH_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)

31
source/libs/scheduler/src/scheduler.c
View File

@ -270,6 +270,8 @@ int32_t schPushTaskToExecList(SSchJob *job, SSchTask *task) {
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SCH_TASK_DLOG("push to %s list", "execTasks");
return TSDB_CODE_SUCCESS;
}
@ -284,6 +286,8 @@ int32_t schMoveTaskToSuccList(SSchJob *job, SSchTask *task, bool *moved) {
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SCH_TASK_DLOG("push to %s list", "succTasks");
*moved = true;
return TSDB_CODE_SUCCESS;
@ -299,6 +303,8 @@ int32_t schMoveTaskToFailList(SSchJob *job, SSchTask *task, bool *moved) {
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
SCH_TASK_DLOG("push to %s list", "failTasks");
*moved = true;
return TSDB_CODE_SUCCESS;
@ -372,7 +378,7 @@ int32_t schProcessOnTaskSuccess(SSchJob *job, SSchTask *task) {
SCH_ERR_RET(schMoveTaskToSuccList(job, task, &moved));
if (!moved) {
SCH_TASK_ERR_LOG(" task may already moved, status:%d", task->status);
SCH_TASK_ELOG("task may already moved, status:%d", task->status);
return TSDB_CODE_SUCCESS;
}
@ -447,11 +453,11 @@ int32_t schProcessOnTaskFailure(SSchJob *job, SSchTask *task, int32_t errCode) {
SCH_ERR_RET(schTaskCheckAndSetRetry(job, task, errCode, &needRetry));
if (!needRetry) {
SCH_TASK_ERR_LOG("task failed[%x], no more retry", errCode);
SCH_TASK_ELOG("task failed[%x], no more retry", errCode);
SCH_ERR_RET(schMoveTaskToFailList(job, task, &moved));
if (!moved) {
SCH_TASK_ERR_LOG("task may already moved, status:%d", task->status);
SCH_TASK_ELOG("task may already moved, status:%d", task->status);
}
if (SCH_TASK_NEED_WAIT_ALL(task)) {
@ -491,7 +497,6 @@ int32_t schProcessRspMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *ms
goto _task_error;
}
}
break;
}
case TDMT_VND_SUBMIT_RSP: {
@ -568,19 +573,25 @@ int32_t schHandleCallback(void* param, const SDataBuf* pMsg, int32_t msgType, in
int32_t code = 0;
SSchCallbackParam *pParam = (SSchCallbackParam *)param;
SSchJob **job = taosHashGet(schMgmt.jobs, &pParam->queryId, sizeof(pParam->queryId));
if (NULL == job || NULL == (*job)) {
SSchJob **pjob = taosHashGet(schMgmt.jobs, &pParam->queryId, sizeof(pParam->queryId));
if (NULL == pjob || NULL == (*pjob)) {
qError("taosHashGet queryId:%"PRIx64" not exist", pParam->queryId);
SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
}
SSchTask **task = taosHashGet((*job)->execTasks, &pParam->taskId, sizeof(pParam->taskId));
if (NULL == task || NULL == (*task)) {
SSchJob *job = *pjob;
SSchTask **ptask = taosHashGet(job->execTasks, &pParam->taskId, sizeof(pParam->taskId));
if (NULL == ptask || NULL == (*ptask)) {
qError("taosHashGet taskId:%"PRIx64" not exist", pParam->taskId);
SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
}
SSchTask *task = *ptask;
SCH_TASK_DLOG("Got msg:%d, rspCode:%d", msgType, rspCode);
schProcessRspMsg(*job, *task, msgType, pMsg->pData, pMsg->len, rspCode);
schProcessRspMsg(job, task, msgType, pMsg->pData, pMsg->len, rspCode);
_return:
tfree(param);
@ -810,7 +821,7 @@ int32_t schLaunchTask(SSchJob *job, SSchTask *task) {
SCH_ERR_RET(schSetTaskCondidateAddrs(job, task));
if (NULL == task->condidateAddrs || taosArrayGetSize(task->condidateAddrs) <= 0) {
SCH_TASK_ERR_LOG("no valid condidate node for task:%"PRIx64, task->taskId);
SCH_TASK_ELOG("no valid condidate node for task:%"PRIx64, task->taskId);
SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
}