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

Merge pull request #1374 from taosdata/refact/slguan 

Refact/slguan
This commit is contained in:
hzcheng 2020-03-15 21:56:39 +08:00 committed by GitHub
parent 86e66e86b1 b23c4cd1c2
commit d6c8f59325
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 182 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/dnode/src/dnodeMgmt.c
View File

@ -185,9 +185,10 @@ static int32_t dnodeOpenVnodes() {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
typedef void (*CleanupFp)(char *);
static void dnodeCleanupVnodes() { static void dnodeCleanupVnodes() {
int32_t num = taosGetIntHashSize(tsDnodeVnodesHash); int32_t num = taosGetIntHashSize(tsDnodeVnodesHash);
taosCleanUpIntHashWithFp(tsDnodeVnodesHash, dnodeCleanupVnode); taosCleanUpIntHashWithFp(tsDnodeVnodesHash, (CleanupFp)dnodeCleanupVnode);
dPrint("all vnodes is opened, num:%d", num); dPrint("all vnodes is opened, num:%d", num);
} }
@ -207,7 +208,7 @@ static int32_t dnodeOpenVnode(int32_t vnode, char *rootDir) {
vnodeObj.vnode = vnode; //tsdbInfo->tsdbCfg.tsdbId; vnodeObj.vnode = vnode; //tsdbInfo->tsdbCfg.tsdbId;
vnodeObj.status = TSDB_VN_STATUS_NOT_READY; vnodeObj.status = TSDB_VN_STATUS_NOT_READY;
vnodeObj.refCount = 1; vnodeObj.refCount = 1;
vnodeObj.version = version; vnodeObj.version = 0;
vnodeObj.wworker = dnodeAllocateWriteWorker(); vnodeObj.wworker = dnodeAllocateWriteWorker();
vnodeObj.rworker = dnodeAllocateReadWorker(); vnodeObj.rworker = dnodeAllocateReadWorker();
vnodeObj.wal = NULL; vnodeObj.wal = NULL;
@ -289,7 +290,7 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
void *pTsdb = tsdbCreateRepo(rootDir, &tsdbCfg, NULL); void *pTsdb = tsdbCreateRepo(rootDir, &tsdbCfg, NULL);
if (pTsdb == NULL) { if (pTsdb == NULL) {
dError("failed to create tsdb in vnode:%d, reason:%s", pVnodeCfg->vnode, tstrerror(terrno)); dError("vgroup:%d, failed to create tsdb in vnode:%d, reason:%s", pVnodeCfg->cfg.vgId, pVnodeCfg->vnode, tstrerror(terrno));
return terrno; return terrno;
} }
@ -309,7 +310,7 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj)); taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj));
dPrint("vnode:%d is created", pVnodeCfg->vnode); dPrint("vgroup:%d, vnode:%d is created", pVnodeCfg->cfg.vgId, pVnodeCfg->vnode);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -339,12 +340,12 @@ static void dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions); pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile); pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
dTrace("start to create vnode:%d", pCreate->vnode); dTrace("vgroup:%d, start to create vnode:%d", pCreate->cfg.vgId, pCreate->vnode);
SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId); SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
if (pVnodeObj != NULL) { if (pVnodeObj != NULL) {
rpcRsp.code = TSDB_CODE_SUCCESS; rpcRsp.code = TSDB_CODE_SUCCESS;
dPrint("vnode:%d is already exist", pCreate->vnode); dPrint("vgroup:%d, vnode:%d is already exist", pCreate->cfg.vgId, pCreate->vnode);
} else { } else {
rpcRsp.code = dnodeCreateVnode(pCreate); rpcRsp.code = dnodeCreateVnode(pCreate);
} }

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

@ -291,13 +291,13 @@ 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("start to create table:%s in vgroup:%d", pTable->tableId, pTable->vgId); 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); void *pVnode = dnodeGetVnode(pTable->vgId);
if (pVnode == NULL) { if (pVnode == NULL) {
rpcRsp.code = TSDB_CODE_INVALID_VGROUP_ID; rpcRsp.code = TSDB_CODE_INVALID_VGROUP_ID;
dTrace("failed to create table:%s in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code)); dTrace("table:%s, failed to create in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code));
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
return; return;
} }
@ -306,7 +306,7 @@ static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) {
if (pTsdb == NULL) { if (pTsdb == NULL) {
dnodeReleaseVnode(pVnode); dnodeReleaseVnode(pVnode);
rpcRsp.code = TSDB_CODE_NOT_ACTIVE_VNODE; rpcRsp.code = TSDB_CODE_NOT_ACTIVE_VNODE;
dTrace("failed to create table:%s in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code)); dTrace("table:%s, failed to create in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code));
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
return; return;
} }
@ -322,35 +322,48 @@ static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) {
pTable->createdTime = htobe64(pTable->createdTime); pTable->createdTime = htobe64(pTable->createdTime);
SSchema *pSchema = (SSchema *) pTable->data; 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);
}
STableCfg tCfg; STableCfg tCfg;
tsdbInitTableCfg(&tCfg, pTable->tableType, pTable->uid, pTable->sid); tsdbInitTableCfg(&tCfg, pTable->tableType, pTable->uid, pTable->sid);
STSchema *pDestSchema = tdNewSchema(pTable->numOfColumns); STSchema *pDestSchema = tdNewSchema(pTable->numOfColumns);
for (int i = 0; i < pTable->numOfColumns; i++) { for (int i = 0; i < pTable->numOfColumns; i++) {
tdSchemaAppendCol(pDestSchema, pSchema[i].type, htons(pSchema[i].colId), htons(pSchema[i].bytes)); tdSchemaAppendCol(pDestSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
} }
tsdbTableSetSchema(&tCfg, pDestSchema, false); tsdbTableSetSchema(&tCfg, pDestSchema, false);
if (pTable->numOfTags != NULL) { if (pTable->numOfTags != 0) {
STSchema *pDestTagSchema = tdNewSchema(pTable->numOfTags); STSchema *pDestTagSchema = tdNewSchema(pTable->numOfTags);
for (int i = pTable->numOfColumns; i < pTable->numOfColumns + pTable->numOfTags; i++) { for (int i = pTable->numOfColumns; i < totalCols; i++) {
tdSchemaAppendCol(pDestTagSchema, pSchema[i].type, htons(pSchema[i].colId), htons(pSchema[i].bytes)); tdSchemaAppendCol(pDestTagSchema, pSchema[i].type, pSchema[i].colId, pSchema[i].bytes);
} }
tsdbTableSetSchema(&tCfg, pDestTagSchema, false); tsdbTableSetSchema(&tCfg, pDestTagSchema, false);
}
if (pTable->tableType == TSDB_CHILD_TABLE) {
// TODO: add data row // TODO: add data row
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);
} }
rpcRsp.code = tsdbCreateTable(pTsdb, &tCfg); rpcRsp.code = tsdbCreateTable(pTsdb, &tCfg);
dnodeReleaseVnode(pVnode); dnodeReleaseVnode(pVnode);
if (rpcRsp.code != TSDB_CODE_SUCCESS) { if (rpcRsp.code != TSDB_CODE_SUCCESS) {
dError("failed to create table:%s in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code)); dError("table:%s, failed to create in vgroup:%d, reason:%s", pTable->tableId, pTable->vgId, tstrerror(rpcRsp.code));
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} else { } else {
dTrace("create table:%s in vgroup:%d finished", pTable->tableId, pTable->vgId); dTrace("table:%s, created in dnode", pTable->tableId);
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} }
} }

1
src/inc/mnode.h
View File

@ -97,7 +97,6 @@ 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;

11
src/inc/taosmsg.h
View File

@ -177,15 +177,6 @@ enum _mgmt_table {
#define TSDB_KILL_MSG_LEN 30 #define TSDB_KILL_MSG_LEN 30
typedef enum {
TSDB_TABLE_TYPE_SUPER_TABLE = 0, // super table
TSDB_TABLE_TYPE_CHILD_TABLE = 1, // table created from super table
TSDB_TABLE_TYPE_NORMAL_TABLE = 2, // ordinary table
TSDB_TABLE_TYPE_STREAM_TABLE = 3, // table created from stream computing
TSDB_TABLE_TYPE_MAX = 4
} ETableType;
#define TSDB_VN_READ_ACCCESS ((char)0x1) #define TSDB_VN_READ_ACCCESS ((char)0x1)
#define TSDB_VN_WRITE_ACCCESS ((char)0x2) #define TSDB_VN_WRITE_ACCCESS ((char)0x2)
#define TSDB_VN_ALL_ACCCESS (TSDB_VN_READ_ACCCESS | TSDB_VN_WRITE_ACCCESS) #define TSDB_VN_ALL_ACCCESS (TSDB_VN_READ_ACCCESS | TSDB_VN_WRITE_ACCCESS)
@ -258,11 +249,9 @@ typedef struct {
int32_t sversion; int32_t sversion;
int32_t tagDataLen; int32_t tagDataLen;
int32_t sqlDataLen; int32_t sqlDataLen;
int32_t numOfVPeers;
uint64_t uid; uint64_t uid;
uint64_t superTableUid; uint64_t superTableUid;
uint64_t createdTime; uint64_t createdTime;
SVnodeDesc vpeerDesc[TSDB_MAX_MPEERS];
char tableId[TSDB_TABLE_ID_LEN + 1]; char tableId[TSDB_TABLE_ID_LEN + 1];
char superTableId[TSDB_TABLE_ID_LEN + 1]; char superTableId[TSDB_TABLE_ID_LEN + 1];
char data[]; char data[];

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

@ -38,7 +38,6 @@ int32_t mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter);
void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable); void mgmtAddTableIntoSuperTable(SSuperTableObj *pStable);
void mgmtRemoveTableFromSuperTable(SSuperTableObj *pStable); void mgmtRemoveTableFromSuperTable(SSuperTableObj *pStable);
void mgmtSetTableDirty(STableInfo *pTable, bool isDirty);
SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable); SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable);
SMDDropSTableMsg *mgmtBuildRemoveSuperTableMsg(STableInfo *pTable); SMDDropSTableMsg *mgmtBuildRemoveSuperTableMsg(STableInfo *pTable);

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

@ -276,32 +276,27 @@ static void *mgmtBuildCreateChildTableMsg(SChildTableObj *pTable, SVgObj *pVgrou
int32_t totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags; int32_t totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags;
int32_t contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen; int32_t contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen;
SMDCreateTableMsg *pCreateTable = rpcMallocCont(contLen); SMDCreateTableMsg *pCreate = rpcMallocCont(contLen);
if (pCreateTable == NULL) { if (pCreate == NULL) {
return NULL; return NULL;
} }
memcpy(pCreateTable->tableId, pTable->tableId, TSDB_TABLE_ID_LEN); memcpy(pCreate->tableId, pTable->tableId, TSDB_TABLE_ID_LEN);
memcpy(pCreateTable->superTableId, pTable->superTable->tableId, TSDB_TABLE_ID_LEN); memcpy(pCreate->superTableId, pTable->superTable->tableId, TSDB_TABLE_ID_LEN);
pCreateTable->tableType = pTable->type; pCreate->contLen = htonl(contLen);
pCreateTable->numOfColumns = htons(pTable->superTable->numOfColumns); pCreate->vgId = htonl(pVgroup->vgId);
pCreateTable->numOfTags = htons(pTable->superTable->numOfTags); pCreate->tableType = pTable->type;
pCreateTable->sid = htonl(pTable->sid); pCreate->numOfColumns = htons(pTable->superTable->numOfColumns);
pCreateTable->sversion = htonl(pTable->superTable->sversion); pCreate->numOfTags = htons(pTable->superTable->numOfTags);
pCreateTable->tagDataLen = htonl(tagDataLen); pCreate->sid = htonl(pTable->sid);
pCreateTable->sqlDataLen = 0; pCreate->sversion = htonl(pTable->superTable->sversion);
pCreateTable->contLen = htonl(contLen); pCreate->tagDataLen = htonl(tagDataLen);
pCreateTable->numOfVPeers = htonl(pVgroup->numOfVnodes); pCreate->sqlDataLen = 0;
pCreateTable->uid = htobe64(pTable->uid); pCreate->uid = htobe64(pTable->uid);
pCreateTable->superTableUid = htobe64(pTable->superTable->uid); pCreate->superTableUid = htobe64(pTable->superTable->uid);
pCreateTable->createdTime = htobe64(pTable->createdTime); pCreate->createdTime = htobe64(pTable->createdTime);
for (int i = 0; i < pVgroup->numOfVnodes; ++i) { SSchema *pSchema = (SSchema *) pCreate->data;
pCreateTable->vpeerDesc[i].ip = htonl(pVgroup->vnodeGid[i].ip);
pCreateTable->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
}
SSchema *pSchema = (SSchema *) pCreateTable->data;
memcpy(pSchema, pTable->superTable->schema, totalCols * sizeof(SSchema)); memcpy(pSchema, pTable->superTable->schema, totalCols * sizeof(SSchema));
for (int32_t col = 0; col < totalCols; ++col) { for (int32_t col = 0; col < totalCols; ++col) {
pSchema->bytes = htons(pSchema->bytes); pSchema->bytes = htons(pSchema->bytes);
@ -309,20 +304,20 @@ static void *mgmtBuildCreateChildTableMsg(SChildTableObj *pTable, SVgObj *pVgrou
pSchema++; pSchema++;
} }
memcpy(pCreateTable + sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema), pTagData, tagDataLen); memcpy(pCreate->data + totalCols * sizeof(SSchema), pTagData, tagDataLen);
return pCreateTable; return pCreate;
} }
int32_t mgmtCreateChildTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid, int32_t mgmtCreateChildTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut) { 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; return TSDB_CODE_TOO_MANY_TABLES;
} }
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);
@ -336,32 +331,31 @@ int32_t mgmtCreateChildTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj
} }
strcpy(pTable->tableId, pCreate->tableId); strcpy(pTable->tableId, pCreate->tableId);
strcpy(pTable->superTableId, pSuperTable->tableId); strcpy(pTable->superTableId, pSuperTable->tableId);
pTable->type = TSDB_TABLE_TYPE_CHILD_TABLE; pTable->type = TSDB_CHILD_TABLE;
pTable->createdTime = taosGetTimestampMs();
pTable->superTable = pSuperTable;
pTable->vgId = pVgroup->vgId;
pTable->sid = sid;
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->vgId = pVgroup->vgId;
pTable->createdTime = taosGetTimestampMs();
pTable->superTable = pSuperTable;
if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) { if (sdbInsertRow(tsChildTableSdb, pTable, 0) < 0) {
free(pTable);
mError("table:%s, update sdb error", pCreate->tableId); mError("table:%s, update sdb error", pCreate->tableId);
return TSDB_CODE_SDB_ERROR; return TSDB_CODE_SDB_ERROR;
} }
pTagData += (TSDB_TABLE_ID_LEN + 1); pTagData += (TSDB_TABLE_ID_LEN + 1);
int32_t tagDataLen = contLen - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1; int32_t tagDataLen = contLen - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
*pDCreateOut = mgmtBuildCreateChildTableMsg(pTable, pVgroup, pTagData, tagDataLen); *pMDCreateOut = mgmtBuildCreateChildTableMsg(pTable, pVgroup, pTagData, tagDataLen);
if (*pDCreateOut == NULL) { if (*pMDCreateOut == NULL) {
mError("table:%s, failed to build create table message", pCreate->tableId); mError("table:%s, failed to build create table message", pCreate->tableId);
return TSDB_CODE_SERV_OUT_OF_MEMORY; return TSDB_CODE_SERV_OUT_OF_MEMORY;
} }
*pTableOut = (STableInfo *) pTable; *pTableOut = (STableInfo *) pTable;
mTrace("table:%s, create ctable in vgroup, vgroup:%d sid:%d vnode:%d uid:%" PRIu64 , mTrace("table:%s, create ctable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
pTable->tableId, pVgroup->vgId, sid, pVgroup->vnodeGid[0].vnode, pTable->uid);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

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

@ -301,13 +301,13 @@ static void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable, SVgObj *pVgr
pCreate->vgId = htonl(pVgroup->vgId); pCreate->vgId = htonl(pVgroup->vgId);
pCreate->tableType = pTable->type; pCreate->tableType = pTable->type;
pCreate->numOfColumns = htons(pTable->numOfColumns); pCreate->numOfColumns = htons(pTable->numOfColumns);
pCreate->numOfTags = htons(0); pCreate->numOfTags = 0;
pCreate->sid = htonl(pTable->sid); pCreate->sid = htonl(pTable->sid);
pCreate->sversion = htonl(pTable->sversion); pCreate->sversion = htonl(pTable->sversion);
pCreate->tagDataLen = htonl(0); pCreate->tagDataLen = 0;
pCreate->sqlDataLen = htonl(pTable->sqlLen); pCreate->sqlDataLen = htonl(pTable->sqlLen);
pCreate->uid = htobe64(pTable->uid); pCreate->uid = htobe64(pTable->uid);
pCreate->superTableUid = htobe64(0); pCreate->superTableUid = 0;
pCreate->createdTime = htobe64(pTable->createdTime); pCreate->createdTime = htobe64(pTable->createdTime);
SSchema *pSchema = (SSchema *) pCreate->data; SSchema *pSchema = (SSchema *) pCreate->data;
@ -338,16 +338,16 @@ int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgOb
} }
strcpy(pTable->tableId, pCreate->tableId); strcpy(pTable->tableId, pCreate->tableId);
pTable->type = TSDB_TABLE_TYPE_NORMAL_TABLE; pTable->type = TSDB_NORMAL_TABLE;
pTable->createdTime = taosGetTimestampMs();
pTable->vgId = pVgroup->vgId; pTable->vgId = pVgroup->vgId;
pTable->sid = sid;
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->createdTime = taosGetTimestampMs();
pTable->sversion = 0; pTable->sversion = 0;
pTable->numOfColumns = pCreate->numOfColumns; pTable->numOfColumns = htons(pCreate->numOfColumns);
pTable->sqlLen = pTable->sqlLen; pTable->sqlLen = htons(pCreate->sqlLen);
int32_t numOfCols = pCreate->numOfColumns; int32_t numOfCols = pTable->numOfColumns;
int32_t schemaSize = numOfCols * sizeof(SSchema); int32_t schemaSize = numOfCols * sizeof(SSchema);
pTable->schema = (SSchema *) calloc(1, schemaSize); pTable->schema = (SSchema *) calloc(1, schemaSize);
if (pTable->schema == NULL) { if (pTable->schema == NULL) {
@ -357,40 +357,40 @@ int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgOb
memcpy(pTable->schema, pCreate->schema, numOfCols * sizeof(SSchema)); memcpy(pTable->schema, pCreate->schema, numOfCols * sizeof(SSchema));
pTable->nextColId = 0; pTable->nextColId = 0;
for (int32_t col = 0; col < pCreate->numOfColumns; col++) { for (int32_t col = 0; col < numOfCols; col++) {
SSchema *tschema = (SSchema *) pTable->schema; SSchema *tschema = pTable->schema;
tschema[col].colId = pTable->nextColId++; tschema[col].colId = pTable->nextColId++;
tschema[col].bytes = htons(tschema[col].bytes);
} }
pTable->sqlLen = pCreate->sqlLen;
if (pTable->sqlLen != 0) { if (pTable->sqlLen != 0) {
pTable->type = TSDB_TABLE_TYPE_STREAM_TABLE; pTable->type = TSDB_STREAM_TABLE;
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; return TSDB_CODE_SERV_OUT_OF_MEMORY;
} }
memcpy(pTable->sql, (char *) (pCreate->schema) + numOfCols * sizeof(SSchema), pCreate->sqlLen); memcpy(pTable->sql, (char *) (pCreate->schema) + numOfCols * sizeof(SSchema), pTable->sqlLen);
pTable->sql[pCreate->sqlLen - 1] = 0; pTable->sql[pTable->sqlLen - 1] = 0;
mTrace("table:%s, stream sql len:%d sql:%s", pCreate->tableId, pCreate->sqlLen, pTable->sql); mTrace("table:%s, stream sql len:%d sql:%s", pTable->tableId, pTable->sqlLen, pTable->sql);
} }
if (sdbInsertRow(tsNormalTableSdb, pTable, 0) < 0) { if (sdbInsertRow(tsNormalTableSdb, pTable, 0) < 0) {
mError("table:%s, update sdb error", pCreate->tableId); mError("table:%s, update sdb error", pTable->tableId);
free(pTable);
return TSDB_CODE_SDB_ERROR; return TSDB_CODE_SDB_ERROR;
} }
*pDCreateOut = mgmtBuildCreateNormalTableMsg(pTable, pVgroup); *pDCreateOut = mgmtBuildCreateNormalTableMsg(pTable, pVgroup);
if (*pDCreateOut == NULL) { if (*pDCreateOut == NULL) {
mError("table:%s, failed to build create table message", pCreate->tableId); mError("table:%s, failed to build create table message", pTable->tableId);
sdbDeleteRow(tsNormalTableSdb, pTable);
return TSDB_CODE_SERV_OUT_OF_MEMORY; return TSDB_CODE_SERV_OUT_OF_MEMORY;
} }
*pTableOut = (STableInfo *) pTable; *pTableOut = (STableInfo *) pTable;
mTrace("table:%s, create ntable in vgroup, vgroup:%d sid:%d vnode:%d uid:%" PRIu64 , mTrace("table:%s, create ntable in vgroup, uid:%" PRIu64 , pTable->tableId, pTable->uid);
pTable->tableId, pVgroup->vgId, sid, pVgroup->vnodeGid[0].vnode, pTable->uid);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

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

@ -184,7 +184,7 @@ static void mgmtProcessShowMsg(SQueuedMsg *pMsg) {
} }
if (!tsMgmtShowMetaFp[pShowMsg->type]) { if (!tsMgmtShowMetaFp[pShowMsg->type]) {
mError("show type:%d %s is not support", pShowMsg->type, taosMsg[pShowMsg->type]); mError("show type:%s is not support", taosGetShowTypeStr(pShowMsg->type));
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_OPS_NOT_SUPPORT); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_OPS_NOT_SUPPORT);
return; return;
} }
@ -206,6 +206,7 @@ static void mgmtProcessShowMsg(SQueuedMsg *pMsg) {
mgmtSaveQhandle(pShow); mgmtSaveQhandle(pShow);
pShowRsp->qhandle = htobe64((uint64_t) pShow); pShowRsp->qhandle = htobe64((uint64_t) pShow);
mTrace("show:%p, type:%s, start to get meta", pShow, taosGetShowTypeStr(pShowMsg->type));
int32_t code = (*tsMgmtShowMetaFp[pShowMsg->type])(&pShowRsp->tableMeta, pShow, pMsg->thandle); int32_t code = (*tsMgmtShowMetaFp[pShowMsg->type])(&pShowRsp->tableMeta, pShow, pMsg->thandle);
if (code == 0) { if (code == 0) {
SRpcMsg rpcRsp = { SRpcMsg rpcRsp = {
@ -217,7 +218,7 @@ static void mgmtProcessShowMsg(SQueuedMsg *pMsg) {
}; };
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
} else { } else {
mError("pShow:%p, type:%d %s, failed to get Meta, code:%d", pShow, pShowMsg->type, taosMsg[pShowMsg->type], code); mError("show:%p, type:%s, failed to get meta, reason:%s", pShow, taosGetShowTypeStr(pShowMsg->type), tstrerror(code));
mgmtFreeQhandle(pShow); mgmtFreeQhandle(pShow);
rpcFreeCont(pShowRsp); rpcFreeCont(pShowRsp);
} }

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

@ -65,6 +65,8 @@ static void mgmtSuperTableActionInit() {
mgmtSuperTableActionFp[SDB_TYPE_DECODE] = mgmtSuperTableActionDecode; mgmtSuperTableActionFp[SDB_TYPE_DECODE] = mgmtSuperTableActionDecode;
mgmtSuperTableActionFp[SDB_TYPE_RESET] = mgmtSuperTableActionReset; mgmtSuperTableActionFp[SDB_TYPE_RESET] = mgmtSuperTableActionReset;
mgmtSuperTableActionFp[SDB_TYPE_DESTROY] = mgmtSuperTableActionDestroy; mgmtSuperTableActionFp[SDB_TYPE_DESTROY] = mgmtSuperTableActionDestroy;
mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_METRIC, mgmtGetShowSuperTableMeta);
mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_METRIC, mgmtRetrieveShowSuperTables);
} }
void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) { void *mgmtSuperTableActionReset(void *row, char *str, int32_t size, int32_t *ssize) {
@ -213,14 +215,14 @@ int32_t mgmtCreateSuperTable(SDbObj *pDb, SCMCreateTableMsg *pCreate) {
} }
strcpy(pStable->tableId, pCreate->tableId); strcpy(pStable->tableId, pCreate->tableId);
pStable->type = TSDB_SUPER_TABLE; pStable->type = TSDB_SUPER_TABLE;
pStable->createdTime = taosGetTimestampMs(); pStable->createdTime = taosGetTimestampMs();
pStable->vgId = 0; pStable->vgId = 0;
pStable->sid = 0; pStable->sid = 0;
pStable->uid = (((uint64_t)pStable->createdTime) << 16) + ((uint64_t)sdbGetVersion() & ((1ul << 16) - 1ul)); pStable->uid = (((uint64_t) pStable->createdTime) << 16) + ((uint64_t) sdbGetVersion() & ((1ul << 16) - 1ul));
pStable->sversion = 0; pStable->sversion = 0;
pStable->numOfColumns = pCreate->numOfColumns; pStable->numOfColumns = htons(pCreate->numOfColumns);
pStable->numOfTags = pCreate->numOfTags; pStable->numOfTags = htons(pCreate->numOfTags);
int32_t numOfCols = pCreate->numOfColumns + pCreate->numOfTags; int32_t numOfCols = pCreate->numOfColumns + pCreate->numOfTags;
int32_t schemaSize = numOfCols * sizeof(SSchema); int32_t schemaSize = numOfCols * sizeof(SSchema);
@ -233,16 +235,18 @@ int32_t mgmtCreateSuperTable(SDbObj *pDb, SCMCreateTableMsg *pCreate) {
memcpy(pStable->schema, pCreate->schema, numOfCols * sizeof(SSchema)); memcpy(pStable->schema, pCreate->schema, numOfCols * sizeof(SSchema));
pStable->nextColId = 0; pStable->nextColId = 0;
for (int32_t col = 0; col < pCreate->numOfColumns; col++) { for (int32_t col = 0; col < numOfCols; col++) {
SSchema *tschema = (SSchema *)pStable->schema; SSchema *tschema = pStable->schema;
tschema[col].colId = pStable->nextColId++; tschema[col].colId = pStable->nextColId++;
tschema[col].bytes = htons(tschema[col].bytes);
} }
if (sdbInsertRow(tsSuperTableSdb, pStable, 0) < 0) { if (sdbInsertRow(tsSuperTableSdb, pStable, 0) < 0) {
mError("table:%s, update sdb error", pCreate->tableId); mError("stable:%s, update sdb error", pStable->tableId);
return TSDB_CODE_SDB_ERROR; return TSDB_CODE_SDB_ERROR;
} }
mPrint("stable:%s, is created, tags:%d cols:%d", pStable->tableId, pStable->numOfTags, pStable->numOfColumns);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -500,7 +504,7 @@ static int32_t mgmtGetShowSuperTableMeta(STableMeta *pMeta, SShowObj *pShow, voi
pShow->bytes[cols] = 8; pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP; pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "created_time"); strcpy(pSchema[cols].name, "create time");
pSchema[cols].bytes = htons(pShow->bytes[cols]); pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++; cols++;

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

@ -137,20 +137,9 @@ int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, boo
static void mgmtCreateTable(SVgObj *pVgroup, SQueuedMsg *pMsg) { static void mgmtCreateTable(SVgObj *pVgroup, SQueuedMsg *pMsg) {
SCMCreateTableMsg *pCreate = pMsg->pCont; SCMCreateTableMsg *pCreate = pMsg->pCont;
pCreate->numOfColumns = htons(pCreate->numOfColumns);
pCreate->numOfTags = htons(pCreate->numOfTags);
pCreate->sqlLen = htons(pCreate->sqlLen);
SSchema *pSchema = (SSchema*) pCreate->schema;
for (int32_t i = 0; i < pCreate->numOfColumns + pCreate->numOfTags; ++i) {
pSchema->bytes = htons(pSchema->bytes);
pSchema->colId = i;
pSchema++;
}
int32_t sid = taosAllocateId(pVgroup->idPool); int32_t sid = taosAllocateId(pVgroup->idPool);
if (sid < 0) { if (sid < 0) {
mTrace("thandle:%p, no enough sid in vgroup:%d, start to create a new one", pMsg->thandle, pVgroup->vgId); mTrace("tables:%s, no enough sid in vgroup:%d", pVgroup->vgId);
mgmtCreateVgroup(pMsg); mgmtCreateVgroup(pMsg);
return; return;
} }
@ -160,15 +149,15 @@ static void mgmtCreateTable(SVgObj *pVgroup, SQueuedMsg *pMsg) {
SMDCreateTableMsg *pMDCreate = NULL; SMDCreateTableMsg *pMDCreate = NULL;
if (pCreate->numOfColumns == 0) { if (pCreate->numOfColumns == 0) {
mTrace("thandle:%p, create ctable:%s, vgroup:%d sid:%d ahandle:%p", pMsg->thandle, pCreate->tableId, pVgroup->vgId, sid, pMsg); 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); code = mgmtCreateChildTable(pCreate, pMsg->contLen, pVgroup, sid, &pMDCreate, &pTable);
} else { } else {
mTrace("thandle:%p, create ntable:%s, vgroup:%d sid:%d ahandle:%p", pMsg->thandle, pCreate->tableId, pVgroup->vgId, sid, pMsg); 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); code = mgmtCreateNormalTable(pCreate, pMsg->contLen, pVgroup, sid, &pMDCreate, &pTable);
} }
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
mTrace("thandle:%p, failed to create table:%s in vgroup:%d", pMsg->thandle, pCreate->tableId, pVgroup->vgId); mTrace("table:%s, failed to create in vgroup:%d", pCreate->tableId, pVgroup->vgId);
mgmtSendSimpleResp(pMsg->thandle, code); mgmtSendSimpleResp(pMsg->thandle, code);
return; return;
} }
@ -282,13 +271,13 @@ int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN; pShow->bytes[cols] = TSDB_TABLE_NAME_LEN;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY; pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "table_name"); strcpy(pSchema[cols].name, "table name");
pSchema[cols].bytes = htons(pShow->bytes[cols]); pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++; cols++;
pShow->bytes[cols] = 8; pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP; pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "created_time"); strcpy(pSchema[cols].name, "create time");
pSchema[cols].bytes = htons(pShow->bytes[cols]); pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++; cols++;
@ -300,7 +289,7 @@ int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN; pShow->bytes[cols] = TSDB_TABLE_NAME_LEN;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY; pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "stable"); strcpy(pSchema[cols].name, "super table name");
pSchema[cols].bytes = htons(pShow->bytes[cols]); pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++; cols++;
@ -437,25 +426,20 @@ SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable) {
return pRemove; return pRemove;
} }
void mgmtSetTableDirty(STableInfo *pTable, bool isDirty) {
// TODO: if dirty, delete from sdb
pTable->dirty = isDirty;
}
void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) { void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
if (mgmtCheckRedirect(pMsg->thandle)) return; if (mgmtCheckRedirect(pMsg->thandle)) return;
SCMCreateTableMsg *pCreate = pMsg->pCont; SCMCreateTableMsg *pCreate = pMsg->pCont;
mTrace("thandle:%p, start to create table:%s", pMsg->thandle, pCreate->tableId); mTrace("table:%s, create msg is received from thandle:%p", pCreate->tableId, pMsg->thandle);
if (mgmtCheckExpired()) { if (mgmtCheckExpired()) {
mError("thandle:%p, failed to create table:%s, grant expired", pCreate->tableId); mError("table:%s, failed to create, grant expired", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_GRANT_EXPIRED); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_GRANT_EXPIRED);
return; return;
} }
if (!pMsg->pUser->writeAuth) { if (!pMsg->pUser->writeAuth) {
mError("thandle:%p, failed to create table:%s, no rights", pMsg->thandle, pCreate->tableId); mError("table:%s, failed to create, no rights", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
return; return;
} }
@ -463,14 +447,14 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
SAcctObj *pAcct = pMsg->pUser->pAcct; SAcctObj *pAcct = pMsg->pUser->pAcct;
int32_t code = mgmtCheckTableLimit(pAcct, htons(pCreate->numOfColumns)); int32_t code = mgmtCheckTableLimit(pAcct, htons(pCreate->numOfColumns));
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
mError("thandle:%p, failed to create table:%s, exceed the limit", pMsg->thandle, pCreate->tableId); mError("table:%s, failed to create, exceed the limit", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
return; return;
} }
pMsg->pDb = mgmtGetDb(pCreate->db); pMsg->pDb = mgmtGetDb(pCreate->db);
if (pMsg->pDb == NULL) { if (pMsg->pDb == NULL) {
mError("thandle:%p, failed to create table:%s, db not selected", pMsg->thandle, pCreate->tableId); mError("table:%s, failed to create, db not selected", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_DB_NOT_SELECTED); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_DB_NOT_SELECTED);
return; return;
} }
@ -478,19 +462,18 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
STableInfo *pTable = mgmtGetTable(pCreate->tableId); STableInfo *pTable = mgmtGetTable(pCreate->tableId);
if (pTable != NULL) { if (pTable != NULL) {
if (pCreate->igExists) { if (pCreate->igExists) {
mTrace("thandle:%p, table:%s is already exist", pMsg->thandle, pCreate->tableId); mTrace("table:%s is already exist", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SUCCESS); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SUCCESS);
return; return;
} else { } else {
mError("thandle:%p, failed to create table:%s, table already exist", pMsg->thandle, pCreate->tableId); mError("table:%s, failed to create, table already exist", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_TABLE_ALREADY_EXIST); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_TABLE_ALREADY_EXIST);
return; return;
} }
} }
if (pCreate->numOfTags != 0) { if (pCreate->numOfTags != 0) {
mTrace("thandle:%p, start to create super table:%s, tags:%d columns:%d", mTrace("table:%s, is a super table", pCreate->tableId);
pMsg->thandle, pCreate->tableId, pCreate->numOfTags, pCreate->numOfColumns);
code = mgmtCreateSuperTable(pMsg->pDb, pCreate); code = mgmtCreateSuperTable(pMsg->pDb, pCreate);
mgmtSendSimpleResp(pMsg->thandle, code); mgmtSendSimpleResp(pMsg->thandle, code);
return; return;
@ -498,7 +481,7 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
code = mgmtCheckTimeSeries(pCreate->numOfColumns); code = mgmtCheckTimeSeries(pCreate->numOfColumns);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
mError("thandle:%p, failed to create table:%s, timeseries exceed the limit", pMsg->thandle, pCreate->tableId); mError("table:%s, failed to create, timeseries exceed the limit", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, code); mgmtSendSimpleResp(pMsg->thandle, code);
return; return;
} }
@ -509,10 +492,10 @@ void mgmtProcessCreateTableMsg(SQueuedMsg *pMsg) {
SVgObj *pVgroup = mgmtGetAvailableVgroup(pMsg->pDb); SVgObj *pVgroup = mgmtGetAvailableVgroup(pMsg->pDb);
if (pVgroup == NULL) { if (pVgroup == NULL) {
mTrace("thandle:%p, table:%s start to create a new vgroup", newMsg->thandle, pCreate->tableId); mTrace("table:%s, start to create a new vgroup", pCreate->tableId);
mgmtCreateVgroup(newMsg); mgmtCreateVgroup(newMsg);
} else { } else {
mTrace("thandle:%p, create table:%s in vgroup:%d", newMsg->thandle, pCreate->tableId, pVgroup->vgId); mTrace("table:%s, vgroup:%d is selected", pCreate->tableId, pVgroup->vgId);
mgmtCreateTable(pVgroup, newMsg); mgmtCreateTable(pVgroup, newMsg);
} }
} }
@ -761,19 +744,19 @@ static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
queueMsg->received++; queueMsg->received++;
STableInfo *pTable = queueMsg->ahandle; STableInfo *pTable = queueMsg->ahandle;
mTrace("thandle:%p, create table:%s rsp received, ahandle:%p code:%d received:%d", mTrace("table:%s, create table rsp received, thandle:%p ahandle:%p result:%s", pTable->tableId, queueMsg->thandle,
queueMsg->thandle, pTable->tableId, rpcMsg->handle, rpcMsg->code, queueMsg->received); rpcMsg->handle, tstrerror(rpcMsg->code));
if (rpcMsg->code != TSDB_CODE_SUCCESS) { if (rpcMsg->code != TSDB_CODE_SUCCESS) {
mgmtSetTableDirty(pTable, true); if (pTable->type == TSDB_CHILD_TABLE) {
//sdbDeleteRow(tsVgroupSdb, pVgroup); sdbDeleteRow(tsChildTableSdb, pTable);
} else if (pTable->type == TSDB_NORMAL_TABLE){
sdbDeleteRow(tsNormalTableSdb, pTable);
} else {}
mError("table:%s, failed to create in dnode, reason:%s", pTable->tableId, tstrerror(rpcMsg->code));
mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code); mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);
mError("table:%s, failed to create in dnode, reason:%s, set it dirty", pTable->tableId, tstrerror(rpcMsg->code));
mgmtSetTableDirty(pTable, true);
} else { } else {
mTrace("table:%s, created in dnode", pTable->tableId); mTrace("table:%s, created in dnode", pTable->tableId);
mgmtSetTableDirty(pTable, false);
if (queueMsg->msgType != TSDB_MSG_TYPE_CM_CREATE_TABLE) { if (queueMsg->msgType != TSDB_MSG_TYPE_CM_CREATE_TABLE) {
SQueuedMsg *newMsg = calloc(1, sizeof(SQueuedMsg)); SQueuedMsg *newMsg = calloc(1, sizeof(SQueuedMsg));
newMsg->msgType = queueMsg->msgType; newMsg->msgType = queueMsg->msgType;
@ -783,7 +766,7 @@ static void mgmtProcessCreateTableRsp(SRpcMsg *rpcMsg) {
newMsg->contLen = queueMsg->contLen; newMsg->contLen = queueMsg->contLen;
newMsg->pCont = rpcMallocCont(newMsg->contLen); newMsg->pCont = rpcMallocCont(newMsg->contLen);
memcpy(newMsg->pCont, queueMsg->pCont, newMsg->contLen); memcpy(newMsg->pCont, queueMsg->pCont, newMsg->contLen);
mTrace("table:%s, start to process get meta", pTable->tableId); mTrace("table:%s, start to get meta", pTable->tableId);
mgmtAddToShellQueue(newMsg); mgmtAddToShellQueue(newMsg);
} else { } else {
mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code); mgmtSendSimpleResp(queueMsg->thandle, rpcMsg->code);

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

@ -150,7 +150,7 @@ void mgmtProcessVgTimer(void *handle, void *tmrId) {
void mgmtCreateVgroup(SQueuedMsg *pMsg) { void mgmtCreateVgroup(SQueuedMsg *pMsg) {
SDbObj *pDb = pMsg->pDb; SDbObj *pDb = pMsg->pDb;
if (pDb == NULL) { if (pDb == NULL) {
mError("thandle:%p, failed to create vgroup, db not found", pMsg->thandle); mError("failed to create vgroup, db not found");
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_DB); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_INVALID_DB);
return; return;
} }
@ -159,7 +159,7 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg) {
strcpy(pVgroup->dbName, pDb->name); strcpy(pVgroup->dbName, pDb->name);
pVgroup->numOfVnodes = pDb->cfg.replications; pVgroup->numOfVnodes = pDb->cfg.replications;
if (mgmtAllocVnodes(pVgroup) != 0) { if (mgmtAllocVnodes(pVgroup) != 0) {
mError("thandle:%p, db:%s no enough dnode to alloc %d vnodes", pMsg->thandle, pDb->name, pVgroup->numOfVnodes); mError("db:%s, no enough dnode to alloc %d vnodes to vgroup", pDb->name, pVgroup->numOfVnodes);
free(pVgroup); free(pVgroup);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_ENOUGH_DNODES); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_ENOUGH_DNODES);
return; return;
@ -175,11 +175,9 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg) {
sdbInsertRow(tsVgroupSdb, pVgroup, 0); sdbInsertRow(tsVgroupSdb, pVgroup, 0);
mPrint("thandle:%p, vgroup:%d is created in mnode, db:%s replica:%d", pMsg->thandle, pVgroup->vgId, pDb->name, mPrint("vgroup:%d, is created in mnode, db:%s replica:%d", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes);
pVgroup->numOfVnodes);
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) { for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
mPrint("thandle:%p, vgroup:%d, dnode:%s vnode:%d", pMsg->thandle, pVgroup->vgId, mPrint("vgroup:%d, dnode:%s vnode:%d", pVgroup->vgId, taosIpStr(pVgroup->vnodeGid[i].ip), pVgroup->vnodeGid[i].vnode);
taosIpStr(pVgroup->vnodeGid[i].ip), pVgroup->vnodeGid[i].vnode);
} }
pMsg->ahandle = pVgroup; pMsg->ahandle = pVgroup;
@ -595,7 +593,7 @@ void mgmtSendCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode, SRpcIpSet *ipSet, vo
} }
void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) { void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle) {
mTrace("send create vgroup:%d msg, ahandle:%p", pVgroup->vgId, ahandle); mTrace("vgroup:%d, send create all vnodes msg, ahandle:%p", pVgroup->vgId, ahandle);
for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) { for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip); SRpcIpSet ipSet = mgmtGetIpSetFromIp(pVgroup->vnodeGid[i].ip);
mgmtSendCreateVnodeMsg(pVgroup, pVgroup->vnodeGid[i].vnode, &ipSet, ahandle); mgmtSendCreateVnodeMsg(pVgroup, pVgroup->vnodeGid[i].vnode, &ipSet, ahandle);
@ -613,9 +611,9 @@ static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
} }
SVgObj *pVgroup = queueMsg->ahandle; SVgObj *pVgroup = queueMsg->ahandle;
mTrace("thandle:%p, vgroup:%d create vnode rsp received, ahandle:%p code:%d received:%d successed:%d expected:%d", mTrace("vgroup:%d, create vnode rsp received, result:%s received:%d successed:%d expected:%d, thandle:%p ahandle:%p",
queueMsg->thandle, pVgroup->vgId, rpcMsg->handle, rpcMsg->code, queueMsg->received, queueMsg->successed, pVgroup->vgId, tstrerror(rpcMsg->code), queueMsg->received, queueMsg->successed, queueMsg->expected,
queueMsg->expected); queueMsg->thandle, rpcMsg->handle);
if (queueMsg->received != queueMsg->expected) return; if (queueMsg->received != queueMsg->expected) return;

22
src/util/inc/tstatus.h
View File

@ -94,17 +94,17 @@ enum TSDB_TABLE_STATUS {
TSDB_METER_STATE_DROPPED = 0x18, TSDB_METER_STATE_DROPPED = 0x18,
}; };
const char* taosGetVgroupStatusStr(int32_t vgroupStatus); char* taosGetVgroupStatusStr(int32_t vgroupStatus);
const char* taosGetDbStatusStr(int32_t dbStatus); char* taosGetDbStatusStr(int32_t dbStatus);
const char* taosGetVnodeStatusStr(int32_t vnodeStatus); char* taosGetVnodeStatusStr(int32_t vnodeStatus);
const char* taosGetVnodeSyncStatusStr(int32_t vnodeSyncStatus); char* taosGetVnodeSyncStatusStr(int32_t vnodeSyncStatus);
const char* taosGetVnodeDropStatusStr(int32_t dropping); char* taosGetVnodeDropStatusStr(int32_t dropping);
const char* taosGetDnodeStatusStr(int32_t dnodeStatus); char* taosGetDnodeStatusStr(int32_t dnodeStatus);
const char* taosGetDnodeLbStatusStr(int32_t dnodeBalanceStatus); char* taosGetDnodeLbStatusStr(int32_t dnodeBalanceStatus);
const char* taosGetVgroupLbStatusStr(int32_t vglbStatus); char* taosGetVgroupLbStatusStr(int32_t vglbStatus);
const char* taosGetVnodeStreamStatusStr(int32_t vnodeStreamStatus); char* taosGetVnodeStreamStatusStr(int32_t vnodeStreamStatus);
char* taosGetTableStatusStr(int32_t tableStatus);
const char* taosGetTableStatusStr(int32_t tableStatus); char *taosGetShowTypeStr(int32_t showType);
#ifdef __cplusplus #ifdef __cplusplus
} }

44
src/util/src/tstatus.c
View File

@ -16,7 +16,7 @@
#include "taosmsg.h" #include "taosmsg.h"
#include "tstatus.h" #include "tstatus.h"
const char* taosGetVgroupStatusStr(int32_t vgroupStatus) { char* taosGetVgroupStatusStr(int32_t vgroupStatus) {
switch (vgroupStatus) { switch (vgroupStatus) {
case TSDB_VG_STATUS_READY: return tstrerror(vgroupStatus); case TSDB_VG_STATUS_READY: return tstrerror(vgroupStatus);
case TSDB_VG_STATUS_IN_PROGRESS: return tstrerror(vgroupStatus); case TSDB_VG_STATUS_IN_PROGRESS: return tstrerror(vgroupStatus);
@ -29,7 +29,7 @@ const char* taosGetVgroupStatusStr(int32_t vgroupStatus) {
} }
} }
const char* taosGetDbStatusStr(int32_t dbStatus) { char* taosGetDbStatusStr(int32_t dbStatus) {
switch (dbStatus) { switch (dbStatus) {
case TSDB_DB_STATUS_READY: return "ready"; case TSDB_DB_STATUS_READY: return "ready";
case TSDB_DB_STATUS_DROPPING: return "dropping"; case TSDB_DB_STATUS_DROPPING: return "dropping";
@ -38,7 +38,7 @@ const char* taosGetDbStatusStr(int32_t dbStatus) {
} }
} }
const char* taosGetVnodeStatusStr(int32_t vnodeStatus) { char* taosGetVnodeStatusStr(int32_t vnodeStatus) {
switch (vnodeStatus) { switch (vnodeStatus) {
case TSDB_VN_STATUS_OFFLINE: return "offline"; case TSDB_VN_STATUS_OFFLINE: return "offline";
case TSDB_VN_STATUS_CREATING: return "creating"; case TSDB_VN_STATUS_CREATING: return "creating";
@ -51,7 +51,7 @@ const char* taosGetVnodeStatusStr(int32_t vnodeStatus) {
} }
} }
const char* taosGetVnodeSyncStatusStr(int32_t vnodeSyncStatus) { char* taosGetVnodeSyncStatusStr(int32_t vnodeSyncStatus) {
switch (vnodeSyncStatus) { switch (vnodeSyncStatus) {
case TSDB_VN_SYNC_STATUS_INIT: return "ready"; case TSDB_VN_SYNC_STATUS_INIT: return "ready";
case TSDB_VN_SYNC_STATUS_SYNCING: return "syncing"; case TSDB_VN_SYNC_STATUS_SYNCING: return "syncing";
@ -61,7 +61,7 @@ const char* taosGetVnodeSyncStatusStr(int32_t vnodeSyncStatus) {
} }
} }
const char* taosGetVnodeDropStatusStr(int32_t dropping) { char* taosGetVnodeDropStatusStr(int32_t dropping) {
switch (dropping) { switch (dropping) {
case TSDB_VN_DROP_STATUS_READY: return "ready"; case TSDB_VN_DROP_STATUS_READY: return "ready";
case TSDB_VN_DROP_STATUS_DROPPING: return "dropping"; case TSDB_VN_DROP_STATUS_DROPPING: return "dropping";
@ -69,7 +69,7 @@ const char* taosGetVnodeDropStatusStr(int32_t dropping) {
} }
} }
const char* taosGetDnodeStatusStr(int32_t dnodeStatus) { char* taosGetDnodeStatusStr(int32_t dnodeStatus) {
switch (dnodeStatus) { switch (dnodeStatus) {
case TSDB_DN_STATUS_OFFLINE: return "offline"; case TSDB_DN_STATUS_OFFLINE: return "offline";
case TSDB_DN_STATUS_READY: return "ready"; case TSDB_DN_STATUS_READY: return "ready";
@ -77,7 +77,7 @@ const char* taosGetDnodeStatusStr(int32_t dnodeStatus) {
} }
} }
const char* taosGetDnodeLbStatusStr(int32_t dnodeBalanceStatus) { char* taosGetDnodeLbStatusStr(int32_t dnodeBalanceStatus) {
switch (dnodeBalanceStatus) { switch (dnodeBalanceStatus) {
case TSDB_DN_LB_STATUS_BALANCED: return "balanced"; case TSDB_DN_LB_STATUS_BALANCED: return "balanced";
case TSDB_DN_LB_STATUS_BALANCING: return "balancing"; case TSDB_DN_LB_STATUS_BALANCING: return "balancing";
@ -87,7 +87,7 @@ const char* taosGetDnodeLbStatusStr(int32_t dnodeBalanceStatus) {
} }
} }
const char* taosGetVgroupLbStatusStr(int32_t vglbStatus) { char* taosGetVgroupLbStatusStr(int32_t vglbStatus) {
switch (vglbStatus) { switch (vglbStatus) {
case TSDB_VG_LB_STATUS_READY: return "ready"; case TSDB_VG_LB_STATUS_READY: return "ready";
case TSDB_VG_LB_STATUS_UPDATE: return "updating"; case TSDB_VG_LB_STATUS_UPDATE: return "updating";
@ -95,7 +95,7 @@ const char* taosGetVgroupLbStatusStr(int32_t vglbStatus) {
} }
} }
const char* taosGetVnodeStreamStatusStr(int32_t vnodeStreamStatus) { char* taosGetVnodeStreamStatusStr(int32_t vnodeStreamStatus) {
switch (vnodeStreamStatus) { switch (vnodeStreamStatus) {
case TSDB_VN_STREAM_STATUS_START: return "start"; case TSDB_VN_STREAM_STATUS_START: return "start";
case TSDB_VN_STREAM_STATUS_STOP: return "stop"; case TSDB_VN_STREAM_STATUS_STOP: return "stop";
@ -103,9 +103,9 @@ const char* taosGetVnodeStreamStatusStr(int32_t vnodeStreamStatus) {
} }
} }
const char* taosGetTableStatusStr(int32_t tableStatus) { char* taosGetTableStatusStr(int32_t tableStatus) {
switch(tableStatus) { switch(tableStatus) {
case TSDB_METER_STATE_INSERTING: return "inserting"; case TSDB_METER_STATE_INSERTING:return "inserting";
case TSDB_METER_STATE_IMPORTING:return "importing"; case TSDB_METER_STATE_IMPORTING:return "importing";
case TSDB_METER_STATE_UPDATING: return "updating"; case TSDB_METER_STATE_UPDATING: return "updating";
case TSDB_METER_STATE_DROPPING: return "deleting"; case TSDB_METER_STATE_DROPPING: return "deleting";
@ -114,3 +114,25 @@ const char* taosGetTableStatusStr(int32_t tableStatus) {
default:return "undefined"; default:return "undefined";
} }
} }
char *taosGetShowTypeStr(int32_t showType) {
switch (showType) {
case TSDB_MGMT_TABLE_ACCT: return "show accounts";
case TSDB_MGMT_TABLE_USER: return "show users";
case TSDB_MGMT_TABLE_DB: return "show databases";
case TSDB_MGMT_TABLE_TABLE: return "show tables";
case TSDB_MGMT_TABLE_DNODE: return "show dnodes";
case TSDB_MGMT_TABLE_MNODE: return "show mnodes";
case TSDB_MGMT_TABLE_VGROUP: return "show vgroups";
case TSDB_MGMT_TABLE_METRIC: return "show stables";
case TSDB_MGMT_TABLE_MODULE: return "show modules";
case TSDB_MGMT_TABLE_QUERIES: return "show queries";
case TSDB_MGMT_TABLE_STREAMS: return "show streams";
case TSDB_MGMT_TABLE_CONFIGS: return "show configs";
case TSDB_MGMT_TABLE_CONNS: return "show connections";
case TSDB_MGMT_TABLE_SCORES: return "show scores";
case TSDB_MGMT_TABLE_GRANTS: return "show grants";
case TSDB_MGMT_TABLE_VNODES: return "show vnodes";
default: return "undefined";
}
}

3
src/util/src/tstring.c
View File

@ -128,4 +128,5 @@ char *taosMsg[] = {
"sdb-forward", "sdb-forward",
"sdb-forward-rsp", "sdb-forward-rsp",
"max" "max"
}; };