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

Merge pull request #1381 from taosdata/refact/slguan 

Refact/slguan
This commit is contained in:
hzcheng 2020-03-17 09:29:39 +08:00 committed by GitHub
parent d6c8f59325 562967f1bd
commit 04e0fa8047
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 637 additions and 516 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/client/src/tscServer.c
View File

@ -180,7 +180,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
return TSDB_CODE_CLI_OUT_OF_MEMORY; return TSDB_CODE_CLI_OUT_OF_MEMORY;
} }
pSql->ipList->ip[0] = inet_addr("192.168.0.1"); pSql->ipList->ip[0] = inet_addr(tsPrivateIp);
if (pSql->cmd.command < TSDB_SQL_MGMT) { if (pSql->cmd.command < TSDB_SQL_MGMT) {
pSql->ipList->port = tsDnodeShellPort; pSql->ipList->port = tsDnodeShellPort;
@ -197,7 +197,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
rpcSendRequest(pVnodeConn, pSql->ipList, &rpcMsg); rpcSendRequest(pVnodeConn, pSql->ipList, &rpcMsg);
} else { } else {
pSql->ipList->port = tsMnodeShellPort; pSql->ipList->port = tsMnodeShellPort;
tscPrint("%p msg:%s is sent to server %d", pSql, taosMsg[pSql->cmd.msgType], pSql->ipList->port); tscTrace("%p msg:%s is sent to server %d", pSql, taosMsg[pSql->cmd.msgType], pSql->ipList->port);
memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen); memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.msgType = pSql->cmd.msgType, .msgType = pSql->cmd.msgType,
@ -213,7 +213,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
} }
void tscProcessMsgFromServer(SRpcMsg *rpcMsg) { void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
tscPrint("response:%s is received, len:%d error:%s", taosMsg[rpcMsg->msgType], rpcMsg->contLen, tstrerror(rpcMsg->code)); tscTrace("response:%s is received, len:%d error:%s", taosMsg[rpcMsg->msgType], rpcMsg->contLen, tstrerror(rpcMsg->code));
SSqlObj *pSql = (SSqlObj *)rpcMsg->handle; SSqlObj *pSql = (SSqlObj *)rpcMsg->handle;
if (pSql == NULL || pSql->signature != pSql) { if (pSql == NULL || pSql->signature != pSql) {
tscError("%p sql is already released, signature:%p", pSql, pSql->signature); tscError("%p sql is already released, signature:%p", pSql, pSql->signature);
@ -2032,6 +2032,7 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
tscClearFieldInfo(&pQueryInfo->fieldsInfo); tscClearFieldInfo(&pQueryInfo->fieldsInfo);
msgLen = pMsg - (char*)pCreateTableMsg; msgLen = pMsg - (char*)pCreateTableMsg;
pCreateTableMsg->contLen = htonl(msgLen);
pCmd->payloadLen = msgLen; pCmd->payloadLen = msgLen;
pCmd->msgType = TSDB_MSG_TYPE_CM_CREATE_TABLE; pCmd->msgType = TSDB_MSG_TYPE_CM_CREATE_TABLE;

44
src/dnode/src/dnodeRead.c
View File

@ -70,25 +70,31 @@ void dnodeCleanupRead() {
} }
void dnodeRead(SRpcMsg *pMsg) { void dnodeRead(SRpcMsg *pMsg) {
int32_t queuedMsgNum = 0;
int32_t leftLen = pMsg->contLen; int32_t leftLen = pMsg->contLen;
char *pCont = (char *) pMsg->pCont; char *pCont = (char *) pMsg->pCont;
int32_t contLen = 0;
int32_t numOfVnodes = 0;
int32_t vgId = 0;
SRpcContext *pRpcContext = NULL; SRpcContext *pRpcContext = NULL;
// parse head, get number of vnodes; // SMsgDesc *pDesc = pCont;
if ( numOfVnodes > 1) { // pDesc->numOfVnodes = htonl(pDesc->numOfVnodes);
pRpcContext = calloc(sizeof(SRpcContext), 1); // pCont += sizeof(SMsgDesc);
pRpcContext->numOfVnodes = 1; // if (pDesc->numOfVnodes > 1) {
// pRpcContext = calloc(sizeof(SRpcContext), 1);
// pRpcContext->numOfVnodes = pDesc->numOfVnodes;
// }
if (pMsg->msgType == TSDB_MSG_TYPE_RETRIEVE) {
queuedMsgNum = 0;
} }
while (leftLen > 0) { while (leftLen > 0) {
// todo: parse head, get vgId, contLen SMsgHead *pHead = (SMsgHead *) pCont;
pHead->vgId = 1; //htonl(pHead->vgId);
pHead->contLen = pMsg->contLen; //htonl(pHead->contLen);
// get pVnode from vgId void *pVnode = dnodeGetVnode(pHead->vgId);
void *pVnode = dnodeGetVnode(vgId);
if (pVnode == NULL) { if (pVnode == NULL) {
leftLen -= pHead->contLen;
pCont -= pHead->contLen;
continue; continue;
} }
@ -96,7 +102,7 @@ void dnodeRead(SRpcMsg *pMsg) {
SReadMsg readMsg; SReadMsg readMsg;
readMsg.rpcMsg = *pMsg; readMsg.rpcMsg = *pMsg;
readMsg.pCont = pCont; readMsg.pCont = pCont;
readMsg.contLen = contLen; readMsg.contLen = pHead->contLen;
readMsg.pRpcContext = pRpcContext; readMsg.pRpcContext = pRpcContext;
readMsg.pVnode = pVnode; readMsg.pVnode = pVnode;
@ -104,11 +110,23 @@ void dnodeRead(SRpcMsg *pMsg) {
taosWriteQitem(queue, &readMsg); taosWriteQitem(queue, &readMsg);
// next vnode // next vnode
leftLen -= contLen; leftLen -= pHead->contLen;
pCont -= contLen; pCont -= pHead->contLen;
queuedMsgNum++;
dnodeReleaseVnode(pVnode); dnodeReleaseVnode(pVnode);
} }
if (queuedMsgNum == 0) {
SRpcMsg rpcRsp = {
.handle = pMsg->handle,
.pCont = NULL,
.contLen = 0,
.code = TSDB_CODE_INVALID_VGROUP_ID,
.msgType = 0
};
rpcSendResponse(&rpcRsp);
}
} }
void *dnodeAllocateReadWorker() { void *dnodeAllocateReadWorker() {

165
src/dnode/src/dnodeWrite.c
View File

@ -93,21 +93,18 @@ void dnodeWrite(SRpcMsg *pMsg) {
char *pCont = (char *) pMsg->pCont; char *pCont = (char *) pMsg->pCont;
SRpcContext *pRpcContext = NULL; SRpcContext *pRpcContext = NULL;
int32_t numOfVnodes = 0; if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT || pMsg->msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) {
if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) { SMsgDesc *pDesc = pCont;
// TODO parse head, get number of vnodes; pDesc->numOfVnodes = htonl(pDesc->numOfVnodes);
numOfVnodes = 1; pCont += sizeof(SMsgDesc);
} else { if (pDesc->numOfVnodes > 1) {
numOfVnodes = 1; pRpcContext = calloc(sizeof(SRpcContext), 1);
} pRpcContext->numOfVnodes = pDesc->numOfVnodes;
}
if (numOfVnodes > 1) {
pRpcContext = calloc(sizeof(SRpcContext), 1);
pRpcContext->numOfVnodes = numOfVnodes;
} }
while (leftLen > 0) { while (leftLen > 0) {
SWriteMsgHead *pHead = (SWriteMsgHead *) pCont; SMsgHead *pHead = (SMsgHead *) pCont;
pHead->vgId = htonl(pHead->vgId); pHead->vgId = htonl(pHead->vgId);
pHead->contLen = htonl(pHead->contLen); pHead->contLen = htonl(pHead->contLen);
@ -291,26 +288,86 @@ static void dnodeProcessSubmitMsg(SWriteMsg *pMsg) {
static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) { static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) {
SMDCreateTableMsg *pTable = pMsg->rpcMsg.pCont; SMDCreateTableMsg *pTable = pMsg->rpcMsg.pCont;
dTrace("table:%s, start to create in dnode, vgroup:%d", pTable->tableId, pTable->vgId);
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
void *pVnode = dnodeGetVnode(pTable->vgId);
if (pVnode == NULL) { dTrace("table:%s, start to create in dnode, vgroup:%d", pTable->tableId, pTable->vgId);
rpcRsp.code = TSDB_CODE_INVALID_VGROUP_ID; pTable->numOfColumns = htons(pTable->numOfColumns);
dTrace("table:%s, failed to create in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code)); pTable->numOfTags = htons(pTable->numOfTags);
rpcSendResponse(&rpcRsp); pTable->sid = htonl(pTable->sid);
return; pTable->sversion = htonl(pTable->sversion);
pTable->tagDataLen = htonl(pTable->tagDataLen);
pTable->sqlDataLen = htonl(pTable->sqlDataLen);
pTable->uid = htobe64(pTable->uid);
pTable->superTableUid = htobe64(pTable->superTableUid);
pTable->createdTime = htobe64(pTable->createdTime);
SSchema *pSchema = (SSchema *) pTable->data;
int totalCols = pTable->numOfColumns + pTable->numOfTags;
for (int i = 0; i < totalCols; i++) {
pSchema[i].colId = htons(pSchema[i].colId);
pSchema[i].bytes = htons(pSchema[i].bytes);
} }
void *pTsdb = dnodeGetVnodeTsdb(pVnode); STableCfg tCfg;
if (pTsdb == NULL) { tsdbInitTableCfg(&tCfg, pTable->tableType, pTable->uid, pTable->sid);
dnodeReleaseVnode(pVnode);
rpcRsp.code = TSDB_CODE_NOT_ACTIVE_VNODE; STSchema *pDestSchema = tdNewSchema(pTable->numOfColumns);
dTrace("table:%s, failed to create in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code)); for (int i = 0; i < pTable->numOfColumns; i++) {
rpcSendResponse(&rpcRsp); tdSchemaAppendCol(pDestSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
return; }
tsdbTableSetSchema(&tCfg, pDestSchema, false);
if (pTable->numOfTags != 0) {
STSchema *pDestTagSchema = tdNewSchema(pTable->numOfTags);
for (int i = pTable->numOfColumns; i < totalCols; i++) {
tdSchemaAppendCol(pDestTagSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
}
tsdbTableSetTagSchema(&tCfg, pDestTagSchema, false);
char *pTagData = pTable->data + totalCols * sizeof(SSchema);
int accumBytes = 0;
SDataRow dataRow = tdNewDataRowFromSchema(pDestTagSchema);
for (int i = 0; i < pTable->numOfTags; i++) {
tdAppendColVal(dataRow, pTagData + accumBytes, pDestTagSchema->columns + i);
accumBytes += pSchema[i + pTable->numOfColumns].bytes;
}
tsdbTableSetTagValue(&tCfg, dataRow, false);
} }
void *pTsdb = dnodeGetVnodeTsdb(pMsg->pVnode);
rpcRsp.code = tsdbCreateTable(pTsdb, &tCfg);
dnodeReleaseVnode(pMsg->pVnode);
dTrace("table:%s, create table result:%s", pTable->tableId, tstrerror(rpcRsp.code));
rpcSendResponse(&rpcRsp);
}
static void dnodeProcessDropTableMsg(SWriteMsg *pMsg) {
SMDDropTableMsg *pTable = pMsg->rpcMsg.pCont;
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
dTrace("table:%s, start to drop in dnode, vgroup:%d", pTable->tableId, pTable->vgId);
STableId tableId = {
.uid = htobe64(pTable->uid),
.tid = htonl(pTable->sid)
};
void *pTsdb = dnodeGetVnodeTsdb(pMsg->pVnode);
rpcRsp.code = tsdbDropTable(pTsdb, tableId);
dnodeReleaseVnode(pMsg->pVnode);
dTrace("table:%s, drop table result:%s", pTable->tableId, tstrerror(rpcRsp.code));
rpcSendResponse(&rpcRsp);
}
static void dnodeProcessAlterTableMsg(SWriteMsg *pMsg) {
SMDCreateTableMsg *pTable = pMsg->rpcMsg.pCont;
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
dTrace("table:%s, start to alter in dnode, vgroup:%d", pTable->tableId, pTable->vgId);
pTable->numOfColumns = htons(pTable->numOfColumns); pTable->numOfColumns = htons(pTable->numOfColumns);
pTable->numOfTags = htons(pTable->numOfTags); pTable->numOfTags = htons(pTable->numOfTags);
pTable->sid = htonl(pTable->sid); pTable->sid = htonl(pTable->sid);
@ -344,7 +401,6 @@ static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) {
} }
tsdbTableSetSchema(&tCfg, pDestTagSchema, false); tsdbTableSetSchema(&tCfg, pDestTagSchema, false);
// TODO: add data row
char *pTagData = pTable->data + totalCols * sizeof(SSchema); char *pTagData = pTable->data + totalCols * sizeof(SSchema);
int accumBytes = 0; int accumBytes = 0;
SDataRow dataRow = tdNewDataRowFromSchema(pDestTagSchema); SDataRow dataRow = tdNewDataRowFromSchema(pDestTagSchema);
@ -356,49 +412,30 @@ static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) {
tsdbTableSetTagValue(&tCfg, dataRow, false); tsdbTableSetTagValue(&tCfg, dataRow, false);
} }
rpcRsp.code = tsdbCreateTable(pTsdb, &tCfg); void *pTsdb = dnodeGetVnodeTsdb(pMsg->pVnode);
dnodeReleaseVnode(pVnode);
if (rpcRsp.code != TSDB_CODE_SUCCESS) { rpcRsp.code = tsdbAlterTable(pTsdb, &tCfg);
dError("table:%s, failed to create in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code)); dnodeReleaseVnode(pMsg->pVnode);
rpcSendResponse(&rpcRsp);
} else {
dTrace("table:%s, created in dnode", pTable->tableId);
rpcSendResponse(&rpcRsp);
}
}
static void dnodeProcessDropTableMsg(SWriteMsg *pMsg) { dTrace("table:%s, alter table result:%s", pTable->tableId, tstrerror(rpcRsp.code));
SMDDropTableMsg *pTable = pMsg->rpcMsg.pCont;
dPrint("table:%s, sid:%d is dropped", pTable->tableId, pTable->sid);
// pTable->sid = htonl(pTable->sid);
// pTable->numOfVPeers = htonl(pTable->numOfVPeers);
// pTable->uid = htobe64(pTable->uid);
//
// for (int i = 0; i < pTable->numOfVPeers; ++i) {
// pTable->vpeerDesc[i].ip = htonl(pTable->vpeerDesc[i].ip);
// pTable->vpeerDesc[i].vnode = htonl(pTable->vpeerDesc[i].vnode);
// }
//
// int32_t code = dnodeDropTable(pTable);
//
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
rpcSendResponse(&rpcRsp);
}
static void dnodeProcessAlterTableMsg(SWriteMsg *pMsg) {
SMDCreateTableMsg *pTable = pMsg->rpcMsg.pCont;
dPrint("table:%s, sid:%d is alterd", pTable->tableId, pTable->sid);
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
static void dnodeProcessDropStableMsg(SWriteMsg *pMsg) { static void dnodeProcessDropStableMsg(SWriteMsg *pMsg) {
SMDDropSTableMsg *pTable = pMsg->rpcMsg.pCont; SMDDropSTableMsg *pTable = pMsg->rpcMsg.pCont;
dPrint("stable:%s, is dropped", pTable->tableId);
SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0}; SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
dTrace("stable:%s, start to it drop in dnode, vgroup:%d", pTable->tableId, pTable->vgId);
pTable->uid = htobe64(pTable->uid);
// TODO: drop stable in vvnode
//void *pTsdb = dnodeGetVnodeTsdb(pMsg->pVnode);
//rpcRsp.code = tsdbDropSTable(pTsdb, pTable->uid);
rpcRsp.code = TSDB_CODE_SUCCESS;
dnodeReleaseVnode(pMsg->pVnode);
dTrace("stable:%s, drop stable result:%s", pTable->tableId, tstrerror(rpcRsp.code));
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }

19
src/inc/mnode.h
View File

@ -39,7 +39,7 @@ extern "C" {
#include "ttimer.h" #include "ttimer.h"
#include "tutil.h" #include "tutil.h"
typedef struct { typedef struct {
uint32_t privateIp; uint32_t privateIp;
int32_t sid; int32_t sid;
uint32_t moduleStatus; uint32_t moduleStatus;
@ -77,6 +77,7 @@ extern "C" {
} SDnodeObj; } SDnodeObj;
typedef struct { typedef struct {
int32_t dnodeId;
uint32_t ip; uint32_t ip;
uint32_t publicIp; uint32_t publicIp;
int32_t vnode; int32_t vnode;
@ -97,6 +98,7 @@ struct _vg_obj;
typedef struct SSuperTableObj { typedef struct SSuperTableObj {
char tableId[TSDB_TABLE_ID_LEN + 1]; char tableId[TSDB_TABLE_ID_LEN + 1];
int8_t type; int8_t type;
int8_t dirty;
uint64_t uid; uint64_t uid;
int32_t sid; int32_t sid;
int32_t vgId; int32_t vgId;
@ -104,7 +106,7 @@ typedef struct SSuperTableObj {
int32_t sversion; int32_t sversion;
int32_t numOfColumns; int32_t numOfColumns;
int32_t numOfTags; int32_t numOfTags;
int8_t reserved[7]; int8_t reserved[5];
int8_t updateEnd[1]; int8_t updateEnd[1];
int32_t numOfTables; int32_t numOfTables;
int16_t nextColId; int16_t nextColId;
@ -114,12 +116,13 @@ typedef struct SSuperTableObj {
typedef struct { typedef struct {
char tableId[TSDB_TABLE_ID_LEN + 1]; char tableId[TSDB_TABLE_ID_LEN + 1];
int8_t type; int8_t type;
int8_t dirty;
uint64_t uid; uint64_t uid;
int32_t sid; int32_t sid;
int32_t vgId; int32_t vgId;
int64_t createdTime; int64_t createdTime;
char superTableId[TSDB_TABLE_ID_LEN + 1]; char superTableId[TSDB_TABLE_ID_LEN + 1];
int8_t reserved[7]; int8_t reserved[1];
int8_t updateEnd[1]; int8_t updateEnd[1];
SSuperTableObj *superTable; SSuperTableObj *superTable;
} SChildTableObj; } SChildTableObj;
@ -127,13 +130,14 @@ typedef struct {
typedef struct { typedef struct {
char tableId[TSDB_TABLE_ID_LEN + 1]; char tableId[TSDB_TABLE_ID_LEN + 1];
int8_t type; int8_t type;
int8_t dirty;
uint64_t uid; uint64_t uid;
int32_t sid; int32_t sid;
int32_t vgId; int32_t vgId;
int64_t createdTime; int64_t createdTime;
int32_t sversion; int32_t sversion;
int32_t numOfColumns; int32_t numOfColumns;
int16_t sqlLen; int32_t sqlLen;
int8_t reserved[3]; int8_t reserved[3];
int8_t updateEnd[1]; int8_t updateEnd[1];
char* sql; //null-terminated string char* sql; //null-terminated string
@ -162,6 +166,7 @@ typedef struct _vg_obj {
typedef struct _db_obj { typedef struct _db_obj {
char name[TSDB_DB_NAME_LEN + 1]; char name[TSDB_DB_NAME_LEN + 1];
int8_t dirty;
int64_t createdTime; int64_t createdTime;
SDbCfg cfg; SDbCfg cfg;
int8_t dropStatus; int8_t dropStatus;
@ -171,10 +176,8 @@ typedef struct _db_obj {
int32_t numOfVgroups; int32_t numOfVgroups;
int32_t numOfTables; int32_t numOfTables;
int32_t numOfSuperTables; int32_t numOfSuperTables;
int32_t vgStatus; SVgObj *pHead;
SVgObj *pHead; // empty vgroup first SVgObj *pTail;
SVgObj *pTail; // empty vgroup end
void * vgTimer;
} SDbObj; } SDbObj;
struct _acctObj; struct _acctObj;

19
src/inc/taosmsg.h
View File

@ -234,10 +234,14 @@ typedef struct {
uint32_t ip; uint32_t ip;
} SVnodeDesc; } SVnodeDesc;
typedef struct {
int32_t numOfVnodes;
} SMsgDesc;
typedef struct { typedef struct {
int32_t contLen; int32_t contLen;
int32_t vgId; int32_t vgId;
} SWriteMsgHead; } SMsgHead;
typedef struct { typedef struct {
int32_t contLen; int32_t contLen;
@ -264,7 +268,8 @@ typedef struct {
int16_t numOfTags; int16_t numOfTags;
int16_t numOfColumns; int16_t numOfColumns;
int16_t sqlLen; // the length of SQL, it starts after schema , sql is a null-terminated string int16_t sqlLen; // the length of SQL, it starts after schema , sql is a null-terminated string
int16_t reserved[16]; int32_t contLen;
int8_t reserved[16];
char schema[]; char schema[];
} SCMCreateTableMsg; } SCMCreateTableMsg;
@ -332,22 +337,22 @@ typedef struct {
} SMgmtHead; } SMgmtHead;
typedef struct { typedef struct {
int32_t contLen;
int32_t vgId; int32_t vgId;
int32_t sid; int32_t sid;
int32_t numOfVPeers;
uint64_t uid; uint64_t uid;
SVnodeDesc vpeerDesc[TSDB_MAX_MPEERS];
char tableId[TSDB_TABLE_ID_LEN + 1]; char tableId[TSDB_TABLE_ID_LEN + 1];
} SMDDropTableMsg; } SMDDropTableMsg;
typedef struct { typedef struct {
char tableId[TSDB_TABLE_ID_LEN + 1]; int32_t contLen;
int32_t vgId;
int64_t uid; int64_t uid;
char tableId[TSDB_TABLE_ID_LEN + 1];
} SMDDropSTableMsg; } SMDDropSTableMsg;
typedef struct { typedef struct {
int32_t vgId; int32_t vgId;
int32_t vnode;
} SMDDropVnodeMsg; } SMDDropVnodeMsg;
typedef struct SColIndexEx { typedef struct SColIndexEx {

9
src/mnode/inc/mgmtChildTable.h
View File

@ -30,13 +30,16 @@ int32_t mgmtInitChildTables();
void mgmtCleanUpChildTables(); void mgmtCleanUpChildTables();
void * mgmtGetChildTable(char *tableId); void * mgmtGetChildTable(char *tableId);
int32_t mgmtCreateChildTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid, void *mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t sid);
SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut); void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pCreate, SChildTableObj *pTable);
int32_t mgmtDropChildTable(SDbObj *pDb, SChildTableObj *pTable);
int32_t mgmtDropChildTable(SQueuedMsg *newMsg, SChildTableObj *pTable);
int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName, char *nContent); int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName, char *nContent);
int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMeta *pMeta, bool usePublicIp); int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMeta *pMeta, bool usePublicIp);
void mgmtDropAllChildTables(SDbObj *pDropDb);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

9
src/mnode/inc/mgmtNormalTable.h
View File

@ -28,14 +28,17 @@ int32_t mgmtInitNormalTables();
void mgmtCleanUpNormalTables(); void mgmtCleanUpNormalTables();
void * mgmtGetNormalTable(char *tableId); void * mgmtGetNormalTable(char *tableId);
int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid, void * mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t sid);
SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut); void * mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable);
int32_t mgmtDropNormalTable(SDbObj *pDb, SNormalTableObj *pTable);
int32_t mgmtDropNormalTable(SQueuedMsg *newMsg, SNormalTableObj *pTable);
int32_t mgmtAddNormalTableColumn(SNormalTableObj *pTable, SSchema schema[], int32_t ncols); int32_t mgmtAddNormalTableColumn(SNormalTableObj *pTable, SSchema schema[], int32_t ncols);
int32_t mgmtDropNormalTableColumnByName(SNormalTableObj *pTable, char *colName); int32_t mgmtDropNormalTableColumnByName(SNormalTableObj *pTable, char *colName);
int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, STableMeta *pMeta, bool usePublicIp); int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, STableMeta *pMeta, bool usePublicIp);
void mgmtDropAllNormalTables(SDbObj *pDropDb);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

6
src/mnode/inc/mgmtSuperTable.h
View File

@ -31,8 +31,8 @@ void mgmtCleanUpSuperTables();
void * mgmtGetSuperTable(char *tableId); void * mgmtGetSuperTable(char *tableId);
int32_t mgmtCreateSuperTable(SDbObj *pDb, SCMCreateTableMsg *pCreate); int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate);
int32_t mgmtDropSuperTable(SDbObj *pDb, SSuperTableObj *pTable); int32_t mgmtDropSuperTable(SQueuedMsg *newMsg, SDbObj *pDb, SSuperTableObj *pTable);
int32_t mgmtAddSuperTableTag(SSuperTableObj *pTable, SSchema schema[], int32_t ntags); int32_t mgmtAddSuperTableTag(SSuperTableObj *pTable, SSchema schema[], int32_t ntags);
int32_t mgmtDropSuperTableTag(SSuperTableObj *pTable, char *tagName); int32_t mgmtDropSuperTableTag(SSuperTableObj *pTable, char *tagName);
int32_t mgmtModifySuperTableTagNameByName(SSuperTableObj *pTable, char *oldTagName, char *newTagName); int32_t mgmtModifySuperTableTagNameByName(SSuperTableObj *pTable, char *oldTagName, char *newTagName);
@ -45,6 +45,8 @@ void * mgmtGetSuperTableVgroup(SSuperTableObj *pStable);
int32_t mgmtFindSuperTableTagIndex(SSuperTableObj *pTable, const char *tagName); int32_t mgmtFindSuperTableTagIndex(SSuperTableObj *pTable, const char *tagName);
int32_t mgmtSetSchemaFromSuperTable(SSchema *pSchema, SSuperTableObj *pTable); int32_t mgmtSetSchemaFromSuperTable(SSchema *pSchema, SSuperTableObj *pTable);
void mgmtDropAllSuperTables(SDbObj *pDropDb);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

7
src/mnode/inc/mgmtTable.h
View File

@ -32,16 +32,9 @@ STableInfo* mgmtGetTable(char *tableId);
STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid); STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid);
int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp); int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp);
int32_t mgmtRetrieveMetricMeta(void *pConn, char **pStart, SSuperTableMetaMsg *pInfo);
int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore);
int32_t mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter);
void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable); void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable);
void mgmtRemoveTableFromSuperTable(SSuperTableObj *pStable); void mgmtRemoveTableFromSuperTable(SSuperTableObj *pStable);
SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable);
SMDDropSTableMsg *mgmtBuildRemoveSuperTableMsg(STableInfo *pTable);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

3
src/mnode/inc/mgmtVgroup.h
View File

@ -30,7 +30,7 @@ SVgObj *mgmtGetVgroup(int32_t vgId);
SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode); SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode);
void mgmtCreateVgroup(SQueuedMsg *pMsg); void mgmtCreateVgroup(SQueuedMsg *pMsg);
int32_t mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup); void mgmtDropVgroup(SVgObj *pVgroup, void *ahandle);
void mgmtUpdateVgroup(SVgObj *pVgroup); void mgmtUpdateVgroup(SVgObj *pVgroup);
void mgmtSetVgroupIdPool(); void mgmtSetVgroupIdPool();
@ -41,7 +41,6 @@ void mgmtRemoveTableFromVgroup(SVgObj *pVgroup, STableInfo *pTable);
SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode); SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode);
void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, void *ahandle); void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, void *ahandle);
void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle);
SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup); SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup);
SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip); SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip);

114
src/mnode/src/mgmtChildTable.c
View File

@ -123,7 +123,6 @@ void *mgmtChildTableActionDelete(void *row, char *str, int32_t size, int32_t *ss
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId); SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) { if (pVgroup == NULL) {
mError("id:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
return NULL; return NULL;
} }
@ -272,19 +271,22 @@ void mgmtCleanUpChildTables() {
sdbCloseTable(tsChildTableSdb); sdbCloseTable(tsChildTableSdb);
} }
static void *mgmtBuildCreateChildTableMsg(SChildTableObj *pTable, SVgObj *pVgroup, void *pTagData, int32_t tagDataLen) { void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableObj *pTable) {
int32_t totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags; char *pTagData = pMsg->schema + TSDB_TABLE_ID_LEN + 1;
int32_t contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen; int32_t tagDataLen = htonl(pMsg->contLen) - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
int32_t totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags;
int32_t contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen;
SMDCreateTableMsg *pCreate = rpcMallocCont(contLen); SMDCreateTableMsg *pCreate = rpcMallocCont(contLen);
if (pCreate == NULL) { if (pCreate == NULL) {
terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
return NULL; return NULL;
} }
memcpy(pCreate->tableId, pTable->tableId, TSDB_TABLE_ID_LEN); memcpy(pCreate->tableId, pTable->tableId, TSDB_TABLE_ID_LEN + 1);
memcpy(pCreate->superTableId, pTable->superTable->tableId, TSDB_TABLE_ID_LEN); memcpy(pCreate->superTableId, pTable->superTable->tableId, TSDB_TABLE_ID_LEN + 1);
pCreate->contLen = htonl(contLen); pCreate->contLen = htonl(contLen);
pCreate->vgId = htonl(pVgroup->vgId); pCreate->vgId = htonl(pTable->vgId);
pCreate->tableType = pTable->type; pCreate->tableType = pTable->type;
pCreate->numOfColumns = htons(pTable->superTable->numOfColumns); pCreate->numOfColumns = htons(pTable->superTable->numOfColumns);
pCreate->numOfTags = htons(pTable->superTable->numOfTags); pCreate->numOfTags = htons(pTable->superTable->numOfTags);
@ -305,104 +307,86 @@ static void *mgmtBuildCreateChildTableMsg(SChildTableObj *pTable, SVgObj *pVgrou
} }
memcpy(pCreate->data + totalCols * sizeof(SSchema), pTagData, tagDataLen); memcpy(pCreate->data + totalCols * sizeof(SSchema), pTagData, tagDataLen);
return pCreate; return pCreate;
} }
int32_t mgmtCreateChildTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid, void* mgmtCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t tid) {
SMDCreateTableMsg **pMDCreateOut, STableInfo **pTableOut) {
int32_t numOfTables = sdbGetNumOfRows(tsChildTableSdb); int32_t numOfTables = sdbGetNumOfRows(tsChildTableSdb);
if (numOfTables >= tsMaxTables) { if (numOfTables >= tsMaxTables) {
mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables); mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
return TSDB_CODE_TOO_MANY_TABLES; terrno = TSDB_CODE_TOO_MANY_TABLES;
return NULL;
} }
char *pTagData = (char *) pCreate->schema; // it is a tag key char *pTagData = (char *) pCreate->schema; // it is a tag key
SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTagData); SSuperTableObj *pSuperTable = mgmtGetSuperTable(pTagData);
if (pSuperTable == NULL) { if (pSuperTable == NULL) {
mError("table:%s, corresponding super table does not exist", pCreate->tableId); mError("table:%s, corresponding super table does not exist", pCreate->tableId);
return TSDB_CODE_INVALID_TABLE; terrno = TSDB_CODE_INVALID_TABLE;
return NULL;
} }
SChildTableObj *pTable = (SChildTableObj *) calloc(sizeof(SChildTableObj), 1); SChildTableObj *pTable = (SChildTableObj *) calloc(sizeof(SChildTableObj), 1);
if (pTable == NULL) { if (pTable == NULL) {
mError("table:%s, failed to alloc memory", pCreate->tableId); mError("table:%s, failed to alloc memory", pCreate->tableId);
return TSDB_CODE_SERV_OUT_OF_MEMORY; terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
return NULL;
} }
strcpy(pTable->tableId, pCreate->tableId); strcpy(pTable->tableId, pCreate->tableId);
strcpy(pTable->superTableId, pSuperTable->tableId); strcpy(pTable->superTableId, pSuperTable->tableId);
pTable->type = TSDB_CHILD_TABLE; pTable->type = TSDB_CHILD_TABLE;
pTable->createdTime = taosGetTimestampMs();
pTable->uid = (((uint64_t) pTable->vgId) << 40) + ((((uint64_t) pTable->sid) & ((1ul << 24) - 1ul)) << 16) + pTable->uid = (((uint64_t) pTable->vgId) << 40) + ((((uint64_t) pTable->sid) & ((1ul << 24) - 1ul)) << 16) +
((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul)); ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
pTable->sid = sid; pTable->sid = tid;
pTable->vgId = pVgroup->vgId; pTable->vgId = pVgroup->vgId;
pTable->createdTime = taosGetTimestampMs();
pTable->superTable = pSuperTable; pTable->superTable = pSuperTable;
if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) { if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) {
free(pTable); free(pTable);
mError("table:%s, update sdb error", pCreate->tableId); mError("table:%s, update sdb error", pCreate->tableId);
return TSDB_CODE_SDB_ERROR; terrno = TSDB_CODE_SDB_ERROR;
return NULL;
} }
pTagData += (TSDB_TABLE_ID_LEN + 1);
int32_t tagDataLen = contLen - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
*pMDCreateOut = mgmtBuildCreateChildTableMsg(pTable, pVgroup, pTagData, tagDataLen);
if (*pMDCreateOut == NULL) {
mError("table:%s, failed to build create table message", pCreate->tableId);
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
*pTableOut = (STableInfo *) pTable;
mTrace("table:%s, create ctable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid); mTrace("table:%s, create ctable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
return TSDB_CODE_SUCCESS; return pTable;
} }
int32_t mgmtDropChildTable(SDbObj *pDb, SChildTableObj *pTable) { int32_t mgmtDropChildTable(SQueuedMsg *newMsg, SChildTableObj *pTable) {
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId); SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) { if (pVgroup == NULL) {
mError("table:%s, failed to drop child table, vgroup not exist", pTable->tableId); mError("table:%s, failed to drop child table, vgroup not exist", pTable->tableId);
return TSDB_CODE_OTHERS; return TSDB_CODE_OTHERS;
} }
SMDDropTableMsg *pRemove = rpcMallocCont(sizeof(SMDDropTableMsg)); SMDDropTableMsg *pDrop = rpcMallocCont(sizeof(SMDDropTableMsg));
if (pRemove == NULL) { if (pDrop == NULL) {
mError("table:%s, failed to drop child table, no enough memory", pTable->tableId); mError("table:%s, failed to drop child table, no enough memory", pTable->tableId);
return TSDB_CODE_SERV_OUT_OF_MEMORY; return TSDB_CODE_SERV_OUT_OF_MEMORY;
} }
strcpy(pRemove->tableId, pTable->tableId); strcpy(pDrop->tableId, pTable->tableId);
pRemove->sid = htonl(pTable->sid); pDrop->vgId = htonl(pTable->vgId);
pRemove->uid = htobe64(pTable->uid); pDrop->contLen = htonl(sizeof(SMDDropTableMsg));
pDrop->sid = htonl(pTable->sid);
pRemove->numOfVPeers = htonl(pVgroup->numOfVnodes); pDrop->uid = htobe64(pTable->uid);
for (int i = 0; i < pVgroup->numOfVnodes; ++i) {
pRemove->vpeerDesc[i].ip = htonl(pVgroup->vnodeGid[i].ip);
pRemove->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
}
SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup); SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup);
mTrace("table:%s, send drop table msg", pRemove->tableId); mTrace("table:%s, send drop table msg", pDrop->tableId);
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.handle = 0, .handle = newMsg,
.pCont = pRemove, .pCont = pDrop,
.contLen = sizeof(SMDDropTableMsg), .contLen = sizeof(SMDDropTableMsg),
.code = 0, .code = 0,
.msgType = TSDB_MSG_TYPE_MD_DROP_TABLE .msgType = TSDB_MSG_TYPE_MD_DROP_TABLE
}; };
newMsg->ahandle = pTable;
mgmtSendMsgToDnode(&ipSet, &rpcMsg); mgmtSendMsgToDnode(&ipSet, &rpcMsg);
if (sdbDeleteRow(tsChildTableSdb, pTable) < 0) {
mError("table:%s, update ctables sdb error", pTable->tableId);
return TSDB_CODE_SDB_ERROR;
}
if (pVgroup->numOfTables <= 0) {
mgmtDropVgroup(pDb, pVgroup);
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -477,13 +461,37 @@ int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMeta *p
for (int32_t i = 0; i < TSDB_VNODES_SUPPORT; ++i) { for (int32_t i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
if (usePublicIp) { if (usePublicIp) {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp; pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
} else { } else {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip; pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
} }
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
pMeta->vpeerDesc[i].vgId = htonl(pVgroup->vgId);
} }
pMeta->numOfVpeers = pVgroup->numOfVnodes; pMeta->numOfVpeers = pVgroup->numOfVnodes;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void mgmtDropAllChildTables(SDbObj *pDropDb) {
void *pNode = NULL;
void *pLastNode = NULL;
int32_t numOfTables = 0;
int32_t dbNameLen = strlen(pDropDb->name);
SChildTableObj *pTable = NULL;
while (1) {
pNode = sdbFetchRow(tsChildTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
sdbDeleteRow(tsChildTableSdb, pTable);
pNode = pLastNode;
numOfTables ++;
continue;
}
}
mTrace("db:%s, all child tables:%d is dropped", pDropDb->name, numOfTables);
}

243
src/mnode/src/mgmtDb.c
View File

@ -27,6 +27,9 @@
#include "mgmtMnode.h" #include "mgmtMnode.h"
#include "mgmtGrant.h" #include "mgmtGrant.h"
#include "mgmtShell.h" #include "mgmtShell.h"
#include "mgmtNormalTable.h"
#include "mgmtChildTable.h"
#include "mgmtSuperTable.h"
#include "mgmtTable.h" #include "mgmtTable.h"
#include "mgmtUser.h" #include "mgmtUser.h"
#include "mgmtVgroup.h" #include "mgmtVgroup.h"
@ -34,10 +37,9 @@
static void *tsDbSdb = NULL; static void *tsDbSdb = NULL;
static int32_t tsDbUpdateSize; static int32_t tsDbUpdateSize;
static int32_t mgmtUpdateDb(SDbObj *pDb);
static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate); static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate);
static int32_t mgmtDropDbByName(SAcctObj *pAcct, char *name, short ignoreNotExists); static void mgmtDropDb(void *handle, void *tmrId);
static int32_t mgmtDropDb(SDbObj *pDb); static void mgmtSetDbDirty(SDbObj *pDb);
static int32_t mgmtGetDbMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetDbMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn);
@ -60,7 +62,7 @@ static void mgmtDbActionInit() {
mgmtDbActionFp[SDB_TYPE_UPDATE] = mgmtDbActionUpdate; mgmtDbActionFp[SDB_TYPE_UPDATE] = mgmtDbActionUpdate;
mgmtDbActionFp[SDB_TYPE_ENCODE] = mgmtDbActionEncode; mgmtDbActionFp[SDB_TYPE_ENCODE] = mgmtDbActionEncode;
mgmtDbActionFp[SDB_TYPE_DECODE] = mgmtDbActionDecode; mgmtDbActionFp[SDB_TYPE_DECODE] = mgmtDbActionDecode;
mgmtDbActionFp[SDB_TYPE_RESET] = mgmtDbActionReset; mgmtDbActionFp[SDB_TYPE_RESET] = mgmtDbActionReset;
mgmtDbActionFp[SDB_TYPE_DESTROY] = mgmtDbActionDestroy; mgmtDbActionFp[SDB_TYPE_DESTROY] = mgmtDbActionDestroy;
} }
@ -98,8 +100,6 @@ int32_t mgmtInitDbs() {
pDb->numOfTables = 0; pDb->numOfTables = 0;
pDb->numOfVgroups = 0; pDb->numOfVgroups = 0;
pDb->numOfSuperTables = 0; pDb->numOfSuperTables = 0;
pDb->vgStatus = TSDB_VG_STATUS_READY;
pDb->vgTimer = NULL;
pAcct = mgmtGetAcct(pDb->cfg.acct); pAcct = mgmtGetAcct(pDb->cfg.acct);
if (pAcct != NULL) if (pAcct != NULL)
mgmtAddDbIntoAcct(pAcct, pDb); mgmtAddDbIntoAcct(pAcct, pDb);
@ -293,135 +293,6 @@ static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate) {
return code; return code;
} }
static int32_t mgmtUpdateDb(SDbObj *pDb) {
return sdbUpdateRow(tsDbSdb, pDb, tsDbUpdateSize, 1);
}
static int32_t mgmtSetDbDropping(SDbObj *pDb) {
if (pDb->dropStatus == TSDB_DB_STATUS_DROP_FROM_SDB) return 0;
SVgObj *pVgroup = pDb->pHead;
while (pVgroup != NULL) {
for (int32_t i = 0; i < pVgroup->numOfVnodes; i++) {
SVnodeGid *pVnodeGid = pVgroup->vnodeGid + i;
SDnodeObj *pDnode = mgmtGetDnode(pVnodeGid->ip);
if (pDnode == NULL) continue;
SVnodeLoad *pVload = &pDnode->vload[pVnodeGid->vnode];
if (pVload->dropStatus != TSDB_VN_DROP_STATUS_DROPPING) {
pVload->dropStatus = TSDB_VN_DROP_STATUS_DROPPING;
mPrint("dnode:%s vnode:%d db:%s set to dropping status", taosIpStr(pDnode->privateIp), pVnodeGid->vnode, pDb->name);
if (mgmtUpdateDnode(pDnode) < 0) {
mError("db:%s drop failed, dnode sdb update error", pDb->name);
return TSDB_CODE_SDB_ERROR;
}
}
}
//void mgmtSendDropVgroupMsg(SVgObj *pVgroup, void *ahandle) {
// mTrace("vgroup:%d send free vgroup msg, ahandle:%p", pVgroup->vgId, ahandle);
//
// for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
// SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
// mgmtSendDropVnodeMsg(pVgroup->vnodeGid[i].vnode, &ipSet, ahandle);
// }
//}
//
// mgmtSendDropVgroupMsg(pVgroup);
pVgroup = pVgroup->next;
}
if (pDb->dropStatus == TSDB_DB_STATUS_DROPPING) return 0;
pDb->dropStatus = TSDB_DB_STATUS_DROPPING;
if (mgmtUpdateDb(pDb) < 0) {
mError("db:%s drop failed, db sdb update error", pDb->name);
return TSDB_CODE_SDB_ERROR;
}
mPrint("db:%s set to dropping status", pDb->name);
return 0;
}
static bool mgmtCheckDropDbFinished(SDbObj *pDb) {
SVgObj *pVgroup = pDb->pHead;
while (pVgroup) {
for (int32_t i = 0; i < pVgroup->numOfVnodes; i++) {
SVnodeGid *pVnodeGid = pVgroup->vnodeGid + i;
SDnodeObj *pDnode = mgmtGetDnode(pVnodeGid->ip);
if (pDnode == NULL) continue;
if (pDnode->status == TSDB_DN_STATUS_OFFLINE) continue;
SVnodeLoad *pVload = &pDnode->vload[pVnodeGid->vnode];
if (pVload->dropStatus == TSDB_VN_DROP_STATUS_DROPPING) {
mTrace("dnode:%s, vnode:%d db:%s wait dropping", taosIpStr(pDnode->privateIp), pVnodeGid->vnode, pDb->name);
return false;
}
}
pVgroup = pVgroup->next;
}
mPrint("db:%s all vnodes drop finished", pDb->name);
return true;
}
static void mgmtDropDbFromSdb(SDbObj *pDb) {
while (pDb->pHead) mgmtDropVgroup(pDb, pDb->pHead);
// SSuperTableObj *pMetric = pDb->pSTable;
// while (pMetric) {
// SSuperTableObj *pNext = pMetric->next;
// mgmtDropTable(pDb, pMetric->tableId, 0);
// pMetric = pNext;
// }
mPrint("db:%s all meters drop finished", pDb->name);
sdbDeleteRow(tsDbSdb, pDb);
mPrint("db:%s database drop finished", pDb->name);
}
static int32_t mgmtDropDb(SDbObj *pDb) {
if (pDb->dropStatus == TSDB_DB_STATUS_DROPPING) {
bool finished = mgmtCheckDropDbFinished(pDb);
if (!finished) {
SVgObj *pVgroup = pDb->pHead;
while (pVgroup != NULL) {
//mgmtSendDropVgroupMsg(pVgroup, NULL);
pVgroup = pVgroup->next;
}
return TSDB_CODE_ACTION_IN_PROGRESS;
}
// don't sync this action
pDb->dropStatus = TSDB_DB_STATUS_DROP_FROM_SDB;
mgmtDropDbFromSdb(pDb);
return 0;
} else {
int32_t code = mgmtSetDbDropping(pDb);
if (code != 0) return code;
return TSDB_CODE_ACTION_IN_PROGRESS;
}
}
UNUSED_FUNC
static int32_t mgmtDropDbByName(SAcctObj *pAcct, char *name, short ignoreNotExists) {
SDbObj *pDb = (SDbObj *)sdbGetRow(tsDbSdb, name);
if (pDb == NULL) {
if (ignoreNotExists) return TSDB_CODE_SUCCESS;
mWarn("db:%s is not there", name);
return TSDB_CODE_INVALID_DB;
}
if (mgmtCheckIsMonitorDB(pDb->name, tsMonitorDbName)) {
return TSDB_CODE_MONITOR_DB_FORBIDDEN;
}
return mgmtDropDb(pDb);
}
bool mgmtCheckIsMonitorDB(char *db, char *monitordb) { bool mgmtCheckIsMonitorDB(char *db, char *monitordb) {
char dbName[TSDB_DB_NAME_LEN + 1] = {0}; char dbName[TSDB_DB_NAME_LEN + 1] = {0};
extractDBName(db, dbName); extractDBName(db, dbName);
@ -430,20 +301,6 @@ bool mgmtCheckIsMonitorDB(char *db, char *monitordb) {
return (strncasecmp(dbName, monitordb, len) == 0 && len == strlen(monitordb)); return (strncasecmp(dbName, monitordb, len) == 0 && len == strlen(monitordb));
} }
UNUSED_FUNC
static void mgmtMonitorDbDrop(void *unused, void *unusedt) {
void * pNode = NULL;
SDbObj *pDb = NULL;
while (1) {
pNode = sdbFetchRow(tsDbSdb, pNode, (void **)&pDb);
if (pDb == NULL) break;
if (pDb->dropStatus != TSDB_DB_STATUS_DROPPING) continue;
mgmtDropDb(pDb);
break;
}
}
static int32_t mgmtAlterDb(SAcctObj *pAcct, SCMAlterDbMsg *pAlter) { static int32_t mgmtAlterDb(SAcctObj *pAcct, SCMAlterDbMsg *pAlter) {
return 0; return 0;
// int32_t code = TSDB_CODE_SUCCESS; // int32_t code = TSDB_CODE_SUCCESS;
@ -840,7 +697,6 @@ void *mgmtDbActionInsert(void *row, char *str, int32_t size, int32_t *ssize) {
pDb->pTail = NULL; pDb->pTail = NULL;
pDb->numOfVgroups = 0; pDb->numOfVgroups = 0;
pDb->numOfTables = 0; pDb->numOfTables = 0;
pDb->vgTimer = NULL;
mgmtAddDbIntoAcct(pAcct, pDb); mgmtAddDbIntoAcct(pAcct, pDb);
return NULL; return NULL;
@ -851,6 +707,10 @@ void *mgmtDbActionDelete(void *row, char *str, int32_t size, int32_t *ssize) {
SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct); SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
mgmtRemoveDbFromAcct(pAcct, pDb); mgmtRemoveDbFromAcct(pAcct, pDb);
mgmtDropAllNormalTables(pDb);
mgmtDropAllChildTables(pDb);
mgmtDropAllSuperTables(pDb);
return NULL; return NULL;
} }
@ -906,6 +766,10 @@ void mgmtRemoveTableFromDb(SDbObj *pDb) {
atomic_add_fetch_32(&pDb->numOfTables, -1); atomic_add_fetch_32(&pDb->numOfTables, -1);
} }
static void mgmtSetDbDirty(SDbObj *pDb) {
pDb->dirty = true;
}
static void mgmtProcessCreateDbMsg(SQueuedMsg *pMsg) { static void mgmtProcessCreateDbMsg(SQueuedMsg *pMsg) {
if (mgmtCheckRedirect(pMsg->thandle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
@ -919,7 +783,6 @@ static void mgmtProcessCreateDbMsg(SQueuedMsg *pMsg) {
pCreate->commitTime = htonl(pCreate->commitTime); pCreate->commitTime = htonl(pCreate->commitTime);
pCreate->blocksPerTable = htons(pCreate->blocksPerTable); pCreate->blocksPerTable = htons(pCreate->blocksPerTable);
pCreate->rowsInFileBlock = htonl(pCreate->rowsInFileBlock); pCreate->rowsInFileBlock = htonl(pCreate->rowsInFileBlock);
// pCreate->cacheNumOfBlocks = htonl(pCreate->cacheNumOfBlocks);
int32_t code; int32_t code;
if (mgmtCheckExpired()) { if (mgmtCheckExpired()) {
@ -957,22 +820,76 @@ static void mgmtProcessAlterDbMsg(SQueuedMsg *pMsg) {
mgmtSendSimpleResp(pMsg->thandle, code); mgmtSendSimpleResp(pMsg->thandle, code);
} }
static void mgmtDropDb(void *handle, void *tmrId) {
SQueuedMsg *newMsg = handle;
SDbObj *pDb = newMsg->ahandle;
mPrint("db:%s, drop db from sdb", pDb->name);
int32_t code = sdbDeleteRow(tsDbSdb, pDb);
if (code != 0) {
code = TSDB_CODE_SDB_ERROR;
}
mgmtSendSimpleResp(newMsg->thandle, code);
rpcFreeCont(newMsg->pCont);
free(newMsg);
}
static void mgmtProcessDropDbMsg(SQueuedMsg *pMsg) { static void mgmtProcessDropDbMsg(SQueuedMsg *pMsg) {
if (mgmtCheckRedirect(pMsg->thandle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
int32_t code; SCMDropDbMsg *pDrop = pMsg->pCont;
if (pMsg->pUser->superAuth) { mTrace("db:%s, drop db msg is received from thandle:%p", pDrop->db, pMsg->thandle);
code = TSDB_CODE_OPS_NOT_SUPPORT;
//SCMDropDbMsg *pDrop = rpcMsg->pCont; if (mgmtCheckExpired()) {
//rpcRsp.code = mgmtDropDbByName(pUser->pAcct, pDrop->db, pDrop->ignoreNotExists); mError("db:%s, failed to drop, grant expired", pDrop->db);
//if (rpcRsp.code == TSDB_CODE_SUCCESS) { mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_GRANT_EXPIRED);
// mLPrint("DB:%s is dropped by %s", pDrop->db, pUser->user); return;
//}
} else {
code = TSDB_CODE_NO_RIGHTS;
} }
if (code != TSDB_CODE_SUCCESS) { if (!pMsg->pUser->writeAuth) {
mgmtSendSimpleResp(pMsg->thandle, code); mError("db:%s, failed to drop, no rights", pDrop->db);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
return;
} }
SDbObj *pDb = mgmtGetDb(pDrop->db);
if (pDb == NULL) {
if (pDrop->ignoreNotExists) {
mTrace("db:%s, db is not exist, think drop success", pDrop->db);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SUCCESS);
return;
} else {
mError("db:%s, failed to drop, invalid db", pDrop->db);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_DB);
return;
}
}
if (mgmtCheckIsMonitorDB(pDb->name, tsMonitorDbName)) {
mError("db:%s, can't drop monitor database", pDrop->db);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_MONITOR_DB_FORBIDDEN);
return;
}
mgmtSetDbDirty(pDb);
SQueuedMsg *newMsg = malloc(sizeof(SQueuedMsg));
memcpy(newMsg, pMsg, sizeof(SQueuedMsg));
pMsg->pCont = NULL;
SVgObj *pVgroup = pDb->pHead;
if (pVgroup != NULL) {
mPrint("vgroup:%d, will be dropped", pVgroup->vgId);
newMsg->ahandle = pVgroup;
newMsg->expected = pVgroup->numOfVnodes;
mgmtDropVgroup(pVgroup, newMsg);
return;
}
mTrace("db:%s, all vgroups is dropped", pDb->name);
void *tmpTmr;
newMsg->ahandle = pDb;
taosTmrReset(mgmtDropDb, 10, newMsg, tsMgmtTmr, &tmpTmr);
} }

101
src/mnode/src/mgmtNormalTable.c
View File

@ -125,7 +125,6 @@ void *mgmtNormalTableActionDelete(void *row, char *str, int32_t size, int32_t *s
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId); SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) { if (pVgroup == NULL) {
mError("id:%s not in vgroup:%d", pTable->tableId, pTable->vgId);
return NULL; return NULL;
} }
@ -287,18 +286,19 @@ void mgmtCleanUpNormalTables() {
sdbCloseTable(tsNormalTableSdb); sdbCloseTable(tsNormalTableSdb);
} }
static void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable, SVgObj *pVgroup) { void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable) {
int32_t totalCols = pTable->numOfColumns; int32_t totalCols = pTable->numOfColumns;
int32_t contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + pTable->sqlLen; int32_t contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + pTable->sqlLen;
SMDCreateTableMsg *pCreate = rpcMallocCont(contLen); SMDCreateTableMsg *pCreate = rpcMallocCont(contLen);
if (pCreate == NULL) { if (pCreate == NULL) {
terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
return NULL; return NULL;
} }
memcpy(pCreate->tableId, pTable->tableId, TSDB_TABLE_ID_LEN + 1); memcpy(pCreate->tableId, pTable->tableId, TSDB_TABLE_ID_LEN + 1);
pCreate->contLen = htonl(contLen); pCreate->contLen = htonl(contLen);
pCreate->vgId = htonl(pVgroup->vgId); pCreate->vgId = htonl(pTable->vgId);
pCreate->tableType = pTable->type; pCreate->tableType = pTable->type;
pCreate->numOfColumns = htons(pTable->numOfColumns); pCreate->numOfColumns = htons(pTable->numOfColumns);
pCreate->numOfTags = 0; pCreate->numOfTags = 0;
@ -319,30 +319,30 @@ static void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable, SVgObj *pVgr
} }
memcpy(pCreate + sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema), pTable->sql, pTable->sqlLen); memcpy(pCreate + sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema), pTable->sql, pTable->sqlLen);
return pCreate; return pCreate;
} }
int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid, void *mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t sid) {
SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut) {
int32_t numOfTables = sdbGetNumOfRows(tsNormalTableSdb); int32_t numOfTables = sdbGetNumOfRows(tsNormalTableSdb);
if (numOfTables >= TSDB_MAX_NORMAL_TABLES) { if (numOfTables >= TSDB_MAX_NORMAL_TABLES) {
mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_NORMAL_TABLES); mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_NORMAL_TABLES);
return TSDB_CODE_TOO_MANY_TABLES; terrno = TSDB_CODE_TOO_MANY_TABLES;
return NULL;
} }
SNormalTableObj *pTable = (SNormalTableObj *) calloc(sizeof(SNormalTableObj), 1); SNormalTableObj *pTable = (SNormalTableObj *) calloc(sizeof(SNormalTableObj), 1);
if (pTable == NULL) { if (pTable == NULL) {
mError("table:%s, failed to alloc memory", pCreate->tableId); mError("table:%s, failed to alloc memory", pCreate->tableId);
return TSDB_CODE_SERV_OUT_OF_MEMORY; terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
return NULL;
} }
strcpy(pTable->tableId, pCreate->tableId); strcpy(pTable->tableId, pCreate->tableId);
pTable->type = TSDB_NORMAL_TABLE; pTable->type = TSDB_NORMAL_TABLE;
pTable->vgId = pVgroup->vgId; pTable->vgId = pVgroup->vgId;
pTable->createdTime = taosGetTimestampMs();
pTable->uid = (((uint64_t) pTable->createdTime) << 16) + ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul)); pTable->uid = (((uint64_t) pTable->createdTime) << 16) + ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
pTable->sid = sid; pTable->sid = sid;
pTable->createdTime = taosGetTimestampMs();
pTable->sversion = 0; pTable->sversion = 0;
pTable->numOfColumns = htons(pCreate->numOfColumns); pTable->numOfColumns = htons(pCreate->numOfColumns);
pTable->sqlLen = htons(pCreate->sqlLen); pTable->sqlLen = htons(pCreate->sqlLen);
@ -352,7 +352,8 @@ int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgOb
pTable->schema = (SSchema *) calloc(1, schemaSize); pTable->schema = (SSchema *) calloc(1, schemaSize);
if (pTable->schema == NULL) { if (pTable->schema == NULL) {
free(pTable); free(pTable);
return TSDB_CODE_SERV_OUT_OF_MEMORY; terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
return NULL;
} }
memcpy(pTable->schema, pCreate->schema, numOfCols * sizeof(SSchema)); memcpy(pTable->schema, pCreate->schema, numOfCols * sizeof(SSchema));
@ -368,7 +369,8 @@ int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgOb
pTable->sql = calloc(1, pTable->sqlLen); pTable->sql = calloc(1, pTable->sqlLen);
if (pTable->sql == NULL) { if (pTable->sql == NULL) {
free(pTable); free(pTable);
return TSDB_CODE_SERV_OUT_OF_MEMORY; terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
return NULL;
} }
memcpy(pTable->sql, (char *) (pCreate->schema) + numOfCols * sizeof(SSchema), pTable->sqlLen); memcpy(pTable->sql, (char *) (pCreate->schema) + numOfCols * sizeof(SSchema), pTable->sqlLen);
pTable->sql[pTable->sqlLen - 1] = 0; pTable->sql[pTable->sqlLen - 1] = 0;
@ -378,65 +380,45 @@ int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgOb
if (sdbInsertRow(tsNormalTableSdb, pTable, 0) < 0) { if (sdbInsertRow(tsNormalTableSdb, pTable, 0) < 0) {
mError("table:%s, update sdb error", pTable->tableId); mError("table:%s, update sdb error", pTable->tableId);
free(pTable); free(pTable);
return TSDB_CODE_SDB_ERROR; terrno = TSDB_CODE_SDB_ERROR;
return NULL;
} }
*pDCreateOut = mgmtBuildCreateNormalTableMsg(pTable, pVgroup);
if (*pDCreateOut == NULL) {
mError("table:%s, failed to build create table message", pTable->tableId);
sdbDeleteRow(tsNormalTableSdb, pTable);
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
*pTableOut = (STableInfo *) pTable;
mTrace("table:%s, create ntable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid); mTrace("table:%s, create ntable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
return TSDB_CODE_SUCCESS; return pTable;
} }
int32_t mgmtDropNormalTable(SDbObj *pDb, SNormalTableObj *pTable) { int32_t mgmtDropNormalTable(SQueuedMsg *newMsg, SNormalTableObj *pTable) {
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId); SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) { if (pVgroup == NULL) {
mError("table:%s, failed to drop normal table, vgroup not exist", pTable->tableId); mError("table:%s, failed to drop normal table, vgroup not exist", pTable->tableId);
return TSDB_CODE_OTHERS; return TSDB_CODE_OTHERS;
} }
SMDDropTableMsg *pRemove = rpcMallocCont(sizeof(SMDDropTableMsg)); SMDDropTableMsg *pDrop = rpcMallocCont(sizeof(SMDDropTableMsg));
if (pRemove == NULL) { if (pDrop == NULL) {
mError("table:%s, failed to drop normal table, no enough memory", pTable->tableId); mError("table:%s, failed to drop normal table, no enough memory", pTable->tableId);
return TSDB_CODE_SERV_OUT_OF_MEMORY; return TSDB_CODE_SERV_OUT_OF_MEMORY;
} }
strcpy(pRemove->tableId, pTable->tableId); strcpy(pDrop->tableId, pTable->tableId);
pRemove->sid = htonl(pTable->sid); pDrop->contLen = htonl(sizeof(SMDDropTableMsg));
pRemove->uid = htobe64(pTable->uid); pDrop->vgId = htonl(pVgroup->vgId);
pDrop->sid = htonl(pTable->sid);
pRemove->numOfVPeers = htonl(pVgroup->numOfVnodes); pDrop->uid = htobe64(pTable->uid);
for (int i = 0; i < pVgroup->numOfVnodes; ++i) {
pRemove->vpeerDesc[i].ip = htonl(pVgroup->vnodeGid[i].ip);
pRemove->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
}
SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup); SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup);
mTrace("table:%s, send drop table msg", pRemove->tableId); mTrace("table:%s, send drop table msg", pDrop->tableId);
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.handle = 0, .handle = newMsg,
.pCont = pRemove, .pCont = pDrop,
.contLen = sizeof(SMDDropTableMsg), .contLen = sizeof(SMDDropTableMsg),
.code = 0, .code = 0,
.msgType = TSDB_MSG_TYPE_MD_DROP_TABLE .msgType = TSDB_MSG_TYPE_MD_DROP_TABLE
}; };
newMsg->ahandle = pTable;
mgmtSendMsgToDnode(&ipSet, &rpcMsg); mgmtSendMsgToDnode(&ipSet, &rpcMsg);
if (sdbDeleteRow(tsNormalTableSdb, pTable) < 0) {
mError("table:%s, update ntables sdb error", pTable->tableId);
return TSDB_CODE_SDB_ERROR;
}
if (pVgroup->numOfTables <= 0) {
mgmtDropVgroup(pDb, pVgroup);
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -557,14 +539,35 @@ int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, STableMeta
for (int32_t i = 0; i < TSDB_VNODES_SUPPORT; ++i) { for (int32_t i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
if (usePublicIp) { if (usePublicIp) {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp; pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
} else { } else {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip; pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
} }
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
pMeta->vpeerDesc[i].vgId = htonl(pVgroup->vgId);
} }
pMeta->numOfVpeers = pVgroup->numOfVnodes; pMeta->numOfVpeers = pVgroup->numOfVnodes;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void mgmtDropAllNormalTables(SDbObj *pDropDb) {
void *pNode = NULL;
void *pLastNode = NULL;
int32_t numOfTables = 0;
int32_t dbNameLen = strlen(pDropDb->name);
SNormalTableObj *pTable = NULL;
while (1) {
pNode = sdbFetchRow(tsNormalTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
sdbDeleteRow(tsNormalTableSdb, pTable);
pNode = pLastNode;
numOfTables ++;
continue;
}
}
mTrace("db:%s, all normal tables:%d is dropped", pDropDb->name, numOfTables);
}

2
src/mnode/src/mgmtShell.c
View File

@ -242,6 +242,8 @@ static void mgmtProcessRetrieveMsg(SQueuedMsg *pMsg) {
} }
SShowObj *pShow = (SShowObj *)pRetrieve->qhandle; SShowObj *pShow = (SShowObj *)pRetrieve->qhandle;
mTrace("show:%p, type:%s, retrieve data", pShow, taosGetShowTypeStr(pShow->type));
if (!mgmtCheckQhandle(pRetrieve->qhandle)) { if (!mgmtCheckQhandle(pRetrieve->qhandle)) {
mError("pShow:%p, query memory is corrupted", pShow); mError("pShow:%p, query memory is corrupted", pShow);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_MEMORY_CORRUPTED); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_MEMORY_CORRUPTED);

41
src/mnode/src/mgmtSuperTable.c
View File

@ -202,7 +202,7 @@ void mgmtCleanUpSuperTables() {
sdbCloseTable(tsSuperTableSdb); sdbCloseTable(tsSuperTableSdb);
} }
int32_t mgmtCreateSuperTable(SDbObj *pDb, SCMCreateTableMsg *pCreate) { int32_t mgmtCreateSuperTable(SCMCreateTableMsg *pCreate) {
int32_t numOfTables = sdbGetNumOfRows(tsSuperTableSdb); int32_t numOfTables = sdbGetNumOfRows(tsSuperTableSdb);
if (numOfTables >= TSDB_MAX_SUPER_TABLES) { if (numOfTables >= TSDB_MAX_SUPER_TABLES) {
mError("stable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_SUPER_TABLES); mError("stable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_SUPER_TABLES);
@ -250,11 +250,16 @@ int32_t mgmtCreateSuperTable(SDbObj *pDb, SCMCreateTableMsg *pCreate) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t mgmtDropSuperTable(SDbObj *pDb, SSuperTableObj *pSuperTable) { int32_t mgmtDropSuperTable(SQueuedMsg *newMsg, SDbObj *pDb, SSuperTableObj *pStable) {
//TODO drop all child tables if (pStable->numOfTables != 0) {
mError("stable:%s, numOfTables:%d not 0", pStable->tableId, pStable->numOfTables);
mgmtRemoveSuperTableFromDb(pDb); return TSDB_CODE_OTHERS;
return sdbDeleteRow(tsSuperTableSdb, pSuperTable); } else {
//TODO: drop child tables
mError("stable:%s, is dropped from sdb", pStable->tableId);
mgmtRemoveSuperTableFromDb(pDb);
return TSDB_CODE_OTHERS;
}
} }
void* mgmtGetSuperTable(char *tableId) { void* mgmtGetSuperTable(char *tableId) {
@ -607,6 +612,30 @@ int32_t mgmtRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows, v
return numOfRows; return numOfRows;
} }
void mgmtDropAllSuperTables(SDbObj *pDropDb) {
void *pNode = NULL;
void *pLastNode = NULL;
int32_t numOfTables = 0;
int32_t dbNameLen = strlen(pDropDb->name);
SSuperTableObj *pTable = NULL;
while (1) {
pNode = sdbFetchRow(tsSuperTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (strncmp(pDropDb->name, pTable->tableId, dbNameLen) == 0) {
sdbDeleteRow(tsSuperTableSdb, pTable);
pNode = pLastNode;
numOfTables ++;
continue;
}
}
mTrace("db:%s, all super tables:%d is dropped", pDropDb->name, numOfTables);
}
void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable) { void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable) {
pStable->numOfTables++; pStable->numOfTables++;
} }

258
src/mnode/src/mgmtTable.c
View File

@ -51,6 +51,7 @@ static void mgmtProcessTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessMultiTableMetaMsg(SQueuedMsg *queueMsg); static void mgmtProcessMultiTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessSuperTableMetaMsg(SQueuedMsg *queueMsg); static void mgmtProcessSuperTableMetaMsg(SQueuedMsg *queueMsg);
static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg); static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg);
static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg);
static int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle); static void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
@ -82,6 +83,9 @@ int32_t mgmtInitTables() {
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_TABLE, mgmtGetShowTableMeta); mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_TABLE, mgmtGetShowTableMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mgmtRetrieveShowTables); mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mgmtRetrieveShowTables);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP, mgmtProcessCreateTableRsp); mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP, mgmtProcessCreateTableRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_TABLE_RSP, mgmtProcessDropTableRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_ALTER_TABLE_RSP, NULL);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_STABLE_RSP, NULL);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -134,83 +138,6 @@ int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, boo
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static void mgmtCreateTable(SVgObj *pVgroup, SQueuedMsg *pMsg) {
SCMCreateTableMsg *pCreate = pMsg->pCont;
int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid < 0) {
mTrace("tables:%s, no enough sid in vgroup:%d", pVgroup->vgId);
mgmtCreateVgroup(pMsg);
return;
}
int32_t code;
STableInfo *pTable;
SMDCreateTableMsg *pMDCreate = NULL;
if (pCreate->numOfColumns == 0) {
mTrace("table:%s, is a child table, vgroup:%d sid:%d ahandle:%p", pCreate->tableId, pVgroup->vgId, sid, pMsg);
code = mgmtCreateChildTable(pCreate, pMsg->contLen, pVgroup, sid, &pMDCreate, &pTable);
} else {
mTrace("table:%s, is a normal table, vgroup:%d sid:%d ahandle:%p", pCreate->tableId, pVgroup->vgId, sid, pMsg);
code = mgmtCreateNormalTable(pCreate, pMsg->contLen, pVgroup, sid, &pMDCreate, &pTable);
}
if (code != TSDB_CODE_SUCCESS) {
mTrace("table:%s, failed to create in vgroup:%d", pCreate->tableId, pVgroup->vgId);
mgmtSendSimpleResp(pMsg->thandle, code);
return;
}
SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup);
SRpcMsg rpcMsg = {
.handle = pMsg,
.pCont = pMDCreate,
.contLen = htonl(pMDCreate->contLen),
.code = 0,
.msgType = TSDB_MSG_TYPE_MD_CREATE_TABLE
};
pMsg->ahandle = pTable;
mgmtSendMsgToDnode(&ipSet, &rpcMsg);
}
int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore) {
STableInfo *pTable = mgmtGetTable(tableId);
if (pTable == NULL) {
if (ignore) {
mTrace("table:%s, table is not exist, think it success", tableId);
return TSDB_CODE_SUCCESS;
} else {
mError("table:%s, failed to create table, table not exist", tableId);
return TSDB_CODE_INVALID_TABLE;
}
}
if (mgmtCheckIsMonitorDB(pDb->name, tsMonitorDbName)) {
mError("table:%s, failed to create table, in monitor database", tableId);
return TSDB_CODE_MONITOR_DB_FORBIDDEN;
}
switch (pTable->type) {
case TSDB_SUPER_TABLE:
mTrace("table:%s, start to drop super table", tableId);
return mgmtDropSuperTable(pDb, (SSuperTableObj *) pTable);
case TSDB_CHILD_TABLE:
mTrace("table:%s, start to drop child table", tableId);
return mgmtDropChildTable(pDb, (SChildTableObj *) pTable);
case TSDB_NORMAL_TABLE:
mTrace("table:%s, start to drop normal table", tableId);
return mgmtDropNormalTable(pDb, (SNormalTableObj *) pTable);
case TSDB_STREAM_TABLE:
mTrace("table:%s, start to drop stream table", tableId);
return mgmtDropNormalTable(pDb, (SNormalTableObj *) pTable);
default:
mError("table:%s, invalid table type:%d", tableId, pTable->type);
return TSDB_CODE_INVALID_TABLE;
}
}
int32_t mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter) { int32_t mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter) {
STableInfo *pTable = mgmtGetTable(pAlter->tableId); STableInfo *pTable = mgmtGetTable(pAlter->tableId);
if (pTable == NULL) { if (pTable == NULL) {
@ -419,13 +346,6 @@ int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *
return numOfRows; return numOfRows;
} }
SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable) {
SMDDropTableMsg *pRemove = NULL;
return pRemove;
}
void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) { void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
if (mgmtCheckRedirect(pMsg->thandle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
@ -474,7 +394,7 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
if (pCreate->numOfTags != 0) { if (pCreate->numOfTags != 0) {
mTrace("table:%s, is a super table", pCreate->tableId); mTrace("table:%s, is a super table", pCreate->tableId);
code = mgmtCreateSuperTable(pMsg->pDb, pCreate); code = mgmtCreateSuperTable(pCreate);
mgmtSendSimpleResp(pMsg->thandle, code); mgmtSendSimpleResp(pMsg->thandle, code);
return; return;
} }
@ -494,29 +414,70 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
if (pVgroup == NULL) { if (pVgroup == NULL) {
mTrace("table:%s, start to create a new vgroup", pCreate->tableId); mTrace("table:%s, start to create a new vgroup", pCreate->tableId);
mgmtCreateVgroup(newMsg); mgmtCreateVgroup(newMsg);
} else { return;
mTrace("table:%s, vgroup:%d is selected", pCreate->tableId, pVgroup->vgId);
mgmtCreateTable(pVgroup, newMsg);
} }
int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid < 0) {
mTrace("tables:%s, no enough sid in vgroup:%d", pVgroup->vgId);
mgmtCreateVgroup(newMsg);
return;
}
SMDCreateTableMsg *pMDCreate = NULL;
if (pCreate->numOfColumns == 0) {
mTrace("table:%s, is a child table, vgroup:%d sid:%d ahandle:%p", pCreate->tableId, pVgroup->vgId, sid, pMsg);
pTable = mgmtCreateChildTable(pCreate, pVgroup, sid);
if (pTable == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
return;
}
pMDCreate = mgmtBuildCreateChildTableMsg(pCreate, (SChildTableObj *) pTable);
if (pMDCreate == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
return;
}
} else {
mTrace("table:%s, is a normal table, vgroup:%d sid:%d ahandle:%p", pCreate->tableId, pVgroup->vgId, sid, pMsg);
pTable = mgmtCreateNormalTable(pCreate, pVgroup, sid);
if (pTable == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
return;
}
pMDCreate = mgmtBuildCreateNormalTableMsg((SNormalTableObj *) pTable);
if (pMDCreate == NULL) {
mgmtSendSimpleResp(pMsg->thandle, terrno);
return;
}
}
SRpcIpSet ipSet = mgmtGetIpSetFromVgroup(pVgroup);
SRpcMsg rpcMsg = {
.handle = newMsg,
.pCont = pMDCreate,
.contLen = htonl(pMDCreate->contLen),
.code = 0,
.msgType = TSDB_MSG_TYPE_MD_CREATE_TABLE
};
newMsg->ahandle = pTable;
mgmtSendMsgToDnode(&ipSet, &rpcMsg);
} }
void mgmtProcessDropTableMsg(SQueuedMsg *pMsg) { void mgmtProcessDropTableMsg(SQueuedMsg *pMsg) {
if (mgmtCheckRedirect(pMsg->thandle)) return;
SCMDropTableMsg *pDrop = pMsg->pCont; SCMDropTableMsg *pDrop = pMsg->pCont;
mTrace("table:%s, drop table msg is received from thandle:%p", pDrop->tableId, pMsg->thandle);
if (mgmtCheckRedirect(pMsg->thandle)) { if (mgmtCheckExpired()) {
mError("thandle:%p, failed to drop table:%s, need redirect message", pMsg->thandle, pDrop->tableId); mError("table:%s, failed to drop, grant expired", pDrop->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_GRANT_EXPIRED);
return; return;
} }
SUserObj *pUser = mgmtGetUserFromConn(pMsg->thandle); if (!pMsg->pUser->writeAuth) {
if (pUser == NULL) { mError("table:%s, failed to drop, no rights", pDrop->tableId);
mError("table:%s, failed to drop table, invalid user", pDrop->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_USER);
return;
}
if (!pUser->writeAuth) {
mError("table:%s, failed to drop table, no rights", pDrop->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
return; return;
} }
@ -528,8 +489,54 @@ void mgmtProcessDropTableMsg(SQueuedMsg *pMsg) {
return; return;
} }
int32_t code = mgmtDropTable(pDb, pDrop->tableId, pDrop->igNotExists); STableInfo *pTable = mgmtGetTable(pDrop->tableId);
if (code != TSDB_CODE_ACTION_IN_PROGRESS) { if (pTable == NULL) {
if (pDrop->igNotExists) {
mTrace("table:%s, table is not exist, think drop success", pDrop->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SUCCESS);
return;
} else {
mError("table:%s, failed to drop table, table not exist", pDrop->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_TABLE);
return;
}
}
if (mgmtCheckIsMonitorDB(pDb->name, tsMonitorDbName)) {
mError("table:%s, failed to drop table, in monitor database", pDrop->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_MONITOR_DB_FORBIDDEN);
return;
}
SQueuedMsg *newMsg = malloc(sizeof(SQueuedMsg));
memcpy(newMsg, pMsg, sizeof(SQueuedMsg));
pMsg->pCont = NULL;
int32_t code;
switch (pTable->type) {
case TSDB_SUPER_TABLE:
mTrace("table:%s, start to drop super table", pDrop->tableId);
code = mgmtDropSuperTable(newMsg, pDb, (SSuperTableObj *) pTable);
break;
case TSDB_CHILD_TABLE:
mTrace("table:%s, start to drop child table", pDrop->tableId);
code = mgmtDropChildTable(newMsg, (SChildTableObj *) pTable);
break;
case TSDB_NORMAL_TABLE:
mTrace("table:%s, start to drop normal table", pDrop->tableId);
code = mgmtDropNormalTable(newMsg, (SNormalTableObj *) pTable);
break;
case TSDB_STREAM_TABLE:
mTrace("table:%s, start to drop stream table", pDrop->tableId);
code = mgmtDropNormalTable(newMsg, (SNormalTableObj *) pTable);
break;
default:
code = TSDB_CODE_INVALID_TABLE_TYPE;
mError("table:%s, invalid table type:%d", pDrop->tableId, pTable->type);
}
if (code != TSDB_CODE_SUCCESS) {
free(newMsg);
mgmtSendSimpleResp(pMsg->thandle, code); mgmtSendSimpleResp(pMsg->thandle, code);
} }
} }
@ -775,3 +782,52 @@ static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
free(queueMsg); free(queueMsg);
} }
static void mgmtProcessDropTableRsp(SRpcMsg *rpcMsg) {
if (rpcMsg->handle == NULL) return;
SQueuedMsg *queueMsg = rpcMsg->handle;
queueMsg->received++;
STableInfo *pTable = queueMsg->ahandle;
mTrace("table:%s, drop table rsp received, thandle:%p result:%s", pTable->tableId, queueMsg->thandle, tstrerror(rpcMsg->code));
if (rpcMsg->code != TSDB_CODE_SUCCESS) {
mError("table:%s, failed to drop in dnode, reason:%s", pTable->tableId, tstrerror(rpcMsg->code));
mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);
free(queueMsg);
return;
}
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {
mError("table:%s, failed to get vgroup", pTable->tableId);
mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_INVALID_VGROUP_ID);
free(queueMsg);
return;
}
if (pTable->type == TSDB_CHILD_TABLE) {
if (sdbDeleteRow(tsChildTableSdb, pTable) < 0) {
mError("table:%s, update ctables sdb error", pTable->tableId);
mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SDB_ERROR);
free(queueMsg);
return;
}
} else if (pTable->type == TSDB_NORMAL_TABLE){
if (sdbDeleteRow(tsNormalTableSdb, pTable) < 0) {
mError("table:%s, update ntables sdb error", pTable->tableId);
mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SDB_ERROR);
free(queueMsg);
return;
}
}
if (pVgroup->numOfTables <= 0) {
mPrint("vgroup:%d, all tables is dropped, drop vgroup", pVgroup->vgId);
mgmtDropVgroup(pVgroup, NULL);
}
mgmtSendSimpleResp(queueMsg->thandle, TSDB_CODE_SUCCESS);
free(queueMsg);
}

104
src/mnode/src/mgmtVgroup.c
View File

@ -44,8 +44,10 @@ static void *mgmtVgroupActionDestroy(void *row, char *str, int32_t size, int32_t
static int32_t mgmtGetVgroupMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn); static int32_t mgmtGetVgroupMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
static int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn); static int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg); static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg);
static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg);
void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle); static void mgmtSendDropVgroupMsg(SVgObj *pVgroup, void *ahandle);
static void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle);
static void mgmtVgroupActionInit() { static void mgmtVgroupActionInit() {
SVgObj tObj; SVgObj tObj;
@ -119,6 +121,7 @@ int32_t mgmtInitVgroups() {
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VGROUP, mgmtGetVgroupMeta); mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_VGROUP, mgmtGetVgroupMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_VGROUP, mgmtRetrieveVgroups); mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_VGROUP, mgmtRetrieveVgroups);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP, mgmtProcessCreateVnodeRsp); mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP, mgmtProcessCreateVnodeRsp);
mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_VNODE_RSP, mgmtProcessDropVnodeRsp);
mTrace("vgroup is initialized"); mTrace("vgroup is initialized");
return 0; return 0;
@ -135,18 +138,6 @@ SVgObj *mgmtGetAvailableVgroup(SDbObj *pDb) {
return pDb->pHead; return pDb->pHead;
} }
void mgmtProcessVgTimer(void *handle, void *tmrId) {
SDbObj *pDb = (SDbObj *)handle;
if (pDb == NULL) return;
if (pDb->vgStatus > TSDB_VG_STATUS_IN_PROGRESS) {
mTrace("db:%s, set vgroup status from %d to ready", pDb->name, pDb->vgStatus);
pDb->vgStatus = TSDB_VG_STATUS_READY;
}
pDb->vgTimer = NULL;
}
void mgmtCreateVgroup(SQueuedMsg *pMsg) { void mgmtCreateVgroup(SQueuedMsg *pMsg) {
SDbObj *pDb = pMsg->pDb; SDbObj *pDb = pMsg->pDb;
if (pDb == NULL) { if (pDb == NULL) {
@ -185,25 +176,14 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg) {
mgmtSendCreateVgroupMsg(pVgroup, pMsg); mgmtSendCreateVgroupMsg(pVgroup, pMsg);
} }
int32_t mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup) { void mgmtDropVgroup(SVgObj *pVgroup, void *ahandle) {
STableInfo *pTable; if (ahandle != NULL) {
mgmtSendDropVgroupMsg(pVgroup, ahandle);
if (pVgroup->numOfTables > 0) { } else {
for (int32_t i = 0; i < pDb->cfg.maxSessions; ++i) { mTrace("vgroup:%d, replica:%d is deleting from sdb", pVgroup->vgId, pVgroup->numOfVnodes);
if (pVgroup->tableList != NULL) { mgmtSendDropVgroupMsg(pVgroup, NULL);
pTable = pVgroup->tableList[i]; sdbDeleteRow(tsVgroupSdb, pVgroup);
if (pTable) mgmtDropTable(pDb, pTable->tableId, 0);
}
}
} }
mTrace("vgroup:%d, db:%s replica:%d is deleted", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes);
//mgmtSendDropVgroupMsg(pVgroup, NULL);
sdbDeleteRow(tsVgroupSdb, pVgroup);
return TSDB_CODE_SUCCESS;
} }
void mgmtSetVgroupIdPool() { void mgmtSetVgroupIdPool() {
@ -634,3 +614,65 @@ static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
free(queueMsg); free(queueMsg);
} }
static SMDDropVnodeMsg *mgmtBuildDropVnodeMsg(SVgObj *pVgroup) {
SMDDropVnodeMsg *pDrop = rpcMallocCont(sizeof(SMDDropVnodeMsg));
if (pDrop == NULL) return NULL;
pDrop->vgId = htonl(pVgroup->vgId);
return pDrop;
}
static void mgmtSendDropVnodeMsg(SVgObj *pVgroup, SRpcIpSet *ipSet, void *ahandle) {
mTrace("vgroup:%d, send drop vnode msg, ahandle:%p", pVgroup->vgId, ahandle);
SMDDropVnodeMsg *pDrop = mgmtBuildDropVnodeMsg(pVgroup);
SRpcMsg rpcMsg = {
.handle = ahandle,
.pCont = pDrop,
.contLen = pDrop ? sizeof(SMDDropVnodeMsg) : 0,
.code = 0,
.msgType = TSDB_MSG_TYPE_MD_DROP_VNODE
};
mgmtSendMsgToDnode(ipSet, &rpcMsg);
}
static void mgmtSendDropVgroupMsg(SVgObj *pVgroup, void *ahandle) {
mTrace("vgroup:%d, send drop all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
mgmtSendDropVnodeMsg(pVgroup, &ipSet, ahandle);
}
}
static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
mTrace("drop vnode msg is received");
if (rpcMsg->handle == NULL) return;
SQueuedMsg *queueMsg = rpcMsg->handle;
queueMsg->received++;
if (rpcMsg->code == TSDB_CODE_SUCCESS) {
queueMsg->code = rpcMsg->code;
queueMsg->successed++;
}
SVgObj *pVgroup = queueMsg->ahandle;
mTrace("vgroup:%d, drop vnode rsp received, result:%s received:%d successed:%d expected:%d, thandle:%p ahandle:%p",
pVgroup->vgId, tstrerror(rpcMsg->code), queueMsg->received, queueMsg->successed, queueMsg->expected,
queueMsg->thandle, rpcMsg->handle);
if (queueMsg->received != queueMsg->expected) return;
sdbDeleteRow(tsVgroupSdb, pVgroup);
SQueuedMsg *newMsg = calloc(1, sizeof(SQueuedMsg));
newMsg->msgType = queueMsg->msgType;
newMsg->thandle = queueMsg->thandle;
newMsg->pDb = queueMsg->pDb;
newMsg->pUser = queueMsg->pUser;
newMsg->contLen = queueMsg->contLen;
newMsg->pCont = rpcMallocCont(newMsg->contLen);
memcpy(newMsg->pCont, queueMsg->pCont, newMsg->contLen);
mgmtAddToShellQueue(newMsg);
free(queueMsg);
}

2
src/vnode/tsdb/CMakeLists.txt
View File

@ -15,5 +15,5 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
TARGET_LINK_LIBRARIES(tsdb common tutil) TARGET_LINK_LIBRARIES(tsdb common tutil)
# Someone has no gtest directory, so comment it # Someone has no gtest directory, so comment it
ADD_SUBDIRECTORY(tests) # ADD_SUBDIRECTORY(tests)
ENDIF () ENDIF ()