TD-1732
This commit is contained in:
Shengliang Guan
parent
e0c0b4f1e1
commit
582cacc5b9
|
|
@ -86,8 +86,8 @@ typedef struct SJoinSupporter {
|
|||
} SJoinSupporter;
|
||||
|
||||
typedef struct SVgroupTableInfo {
|
||||
SCMVgroupInfo vgInfo;
|
||||
SArray* itemList; //SArray<STableIdInfo>
|
||||
SVgroupInfo vgInfo;
|
||||
SArray* itemList; //SArray<STableIdInfo>
|
||||
} SVgroupTableInfo;
|
||||
|
||||
static FORCE_INLINE SQueryInfo* tscGetQueryInfoDetail(SSqlCmd* pCmd, int32_t subClauseIndex) {
|
||||
|
|
@ -237,7 +237,7 @@ void tscDoQuery(SSqlObj* pSql);
|
|||
|
||||
SVgroupsInfo* tscVgroupInfoClone(SVgroupsInfo *pInfo);
|
||||
void* tscVgroupInfoClear(SVgroupsInfo *pInfo);
|
||||
void tscSCMVgroupInfoCopy(SCMVgroupInfo* dst, const SCMVgroupInfo* src);
|
||||
void tscSVgroupInfoCopy(SVgroupInfo* dst, const SVgroupInfo* src);
|
||||
/**
|
||||
* The create object function must be successful expect for the out of memory issue.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -103,7 +103,7 @@ typedef struct STableMeta {
|
|||
int16_t sversion;
|
||||
int16_t tversion;
|
||||
char sTableId[TSDB_TABLE_FNAME_LEN];
|
||||
SCMVgroupInfo vgroupInfo;
|
||||
SVgroupInfo vgroupInfo;
|
||||
SCMCorVgroupInfo corVgroupInfo;
|
||||
STableId id;
|
||||
SSchema schema[]; // if the table is TSDB_CHILD_TABLE, schema is acquired by super table meta info
|
||||
|
|
|
|||
|
|
@ -222,7 +222,7 @@ void tscKillStream(STscObj *pObj, uint32_t killId) {
|
|||
}
|
||||
|
||||
int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
|
||||
SCMHeartBeatMsg *pHeartbeat = pMsg;
|
||||
SHeartBeatMsg *pHeartbeat = pMsg;
|
||||
int allocedQueriesNum = pHeartbeat->numOfQueries;
|
||||
int allocedStreamsNum = pHeartbeat->numOfStreams;
|
||||
|
||||
|
|
@ -277,7 +277,7 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
|
|||
}
|
||||
|
||||
int32_t msgLen = pHeartbeat->numOfQueries * sizeof(SQueryDesc) + pHeartbeat->numOfStreams * sizeof(SStreamDesc) +
|
||||
sizeof(SCMHeartBeatMsg);
|
||||
sizeof(SHeartBeatMsg);
|
||||
pHeartbeat->connId = htonl(pObj->connId);
|
||||
pHeartbeat->numOfQueries = htonl(pHeartbeat->numOfQueries);
|
||||
pHeartbeat->numOfStreams = htonl(pHeartbeat->numOfStreams);
|
||||
|
|
|
|||
|
|
@ -114,7 +114,7 @@ static int32_t optrToString(tSQLExpr* pExpr, char** exprString);
|
|||
static int32_t getTableIndexImpl(SStrToken* pTableToken, SQueryInfo* pQueryInfo, SColumnIndex* pIndex);
|
||||
static int32_t doFunctionsCompatibleCheck(SSqlCmd* pCmd, SQueryInfo* pQueryInfo);
|
||||
static int32_t doLocalQueryProcess(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql);
|
||||
static int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCMCreateDbMsg* pCreate);
|
||||
static int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCreateDbMsg* pCreate);
|
||||
|
||||
static SColumnList getColumnList(int32_t num, int16_t tableIndex, int32_t columnIndex);
|
||||
|
||||
|
|
@ -417,7 +417,7 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
|
|||
|
||||
char* pMsg = pCmd->payload;
|
||||
|
||||
SCMCfgDnodeMsg* pCfg = (SCMCfgDnodeMsg*)pMsg;
|
||||
SCfgDnodeMsg* pCfg = (SCfgDnodeMsg*)pMsg;
|
||||
pDCL->a[0].n = strdequote(pDCL->a[0].z);
|
||||
|
||||
strncpy(pCfg->ep, pDCL->a[0].z, pDCL->a[0].n);
|
||||
|
|
@ -5186,7 +5186,7 @@ int32_t parseLimitClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t clauseIn
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t setKeepOption(SSqlCmd* pCmd, SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
|
||||
static int32_t setKeepOption(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
|
||||
const char* msg = "invalid number of options";
|
||||
|
||||
pMsg->daysToKeep = htonl(-1);
|
||||
|
|
@ -5217,7 +5217,7 @@ static int32_t setKeepOption(SSqlCmd* pCmd, SCMCreateDbMsg* pMsg, SCreateDBInfo*
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t setTimePrecision(SSqlCmd* pCmd, SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDbInfo) {
|
||||
static int32_t setTimePrecision(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDbInfo) {
|
||||
const char* msg = "invalid time precision";
|
||||
|
||||
pMsg->precision = TSDB_TIME_PRECISION_MILLI; // millisecond by default
|
||||
|
|
@ -5241,7 +5241,7 @@ static int32_t setTimePrecision(SSqlCmd* pCmd, SCMCreateDbMsg* pMsg, SCreateDBIn
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static void setCreateDBOption(SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
|
||||
static void setCreateDBOption(SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
|
||||
pMsg->maxTables = htonl(-1); // max tables can not be set anymore
|
||||
pMsg->cacheBlockSize = htonl(pCreateDb->cacheBlockSize);
|
||||
pMsg->totalBlocks = htonl(pCreateDb->numOfBlocks);
|
||||
|
|
@ -5258,7 +5258,7 @@ static void setCreateDBOption(SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
|
|||
}
|
||||
|
||||
int32_t parseCreateDBOptions(SSqlCmd* pCmd, SCreateDBInfo* pCreateDbSql) {
|
||||
SCMCreateDbMsg* pMsg = (SCMCreateDbMsg*)(pCmd->payload);
|
||||
SCreateDbMsg* pMsg = (SCreateDbMsg *)(pCmd->payload);
|
||||
setCreateDBOption(pMsg, pCreateDbSql);
|
||||
|
||||
if (setKeepOption(pCmd, pMsg, pCreateDbSql) != TSDB_CODE_SUCCESS) {
|
||||
|
|
@ -5770,7 +5770,7 @@ int32_t doLocalQueryProcess(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQ
|
|||
}
|
||||
|
||||
// can only perform the parameters based on the macro definitation
|
||||
int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCMCreateDbMsg* pCreate) {
|
||||
int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCreateDbMsg* pCreate) {
|
||||
char msg[512] = {0};
|
||||
|
||||
if (pCreate->walLevel != -1 && (pCreate->walLevel < TSDB_MIN_WAL_LEVEL || pCreate->walLevel > TSDB_MAX_WAL_LEVEL)) {
|
||||
|
|
|
|||
|
|
@ -140,7 +140,7 @@ struct SSchema tscGetTbnameColumnSchema() {
|
|||
strcpy(s.name, TSQL_TBNAME_L);
|
||||
return s;
|
||||
}
|
||||
static void tscInitCorVgroupInfo(SCMCorVgroupInfo *corVgroupInfo, SCMVgroupInfo *vgroupInfo) {
|
||||
static void tscInitCorVgroupInfo(SCMCorVgroupInfo *corVgroupInfo, SVgroupInfo *vgroupInfo) {
|
||||
corVgroupInfo->version = 0;
|
||||
corVgroupInfo->inUse = 0;
|
||||
corVgroupInfo->numOfEps = vgroupInfo->numOfEps;
|
||||
|
|
@ -166,7 +166,7 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size
|
|||
pTableMeta->id.tid = pTableMetaMsg->tid;
|
||||
pTableMeta->id.uid = pTableMetaMsg->uid;
|
||||
|
||||
SCMVgroupInfo* pVgroupInfo = &pTableMeta->vgroupInfo;
|
||||
SVgroupInfo* pVgroupInfo = &pTableMeta->vgroupInfo;
|
||||
pVgroupInfo->numOfEps = pTableMetaMsg->vgroup.numOfEps;
|
||||
pVgroupInfo->vgId = pTableMetaMsg->vgroup.vgId;
|
||||
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ static int32_t getWaitingTimeInterval(int32_t count) {
|
|||
return initial * (2<<(count - 2));
|
||||
}
|
||||
|
||||
static void tscSetDnodeEpSet(SSqlObj* pSql, SCMVgroupInfo* pVgroupInfo) {
|
||||
static void tscSetDnodeEpSet(SSqlObj* pSql, SVgroupInfo* pVgroupInfo) {
|
||||
assert(pSql != NULL && pVgroupInfo != NULL && pVgroupInfo->numOfEps > 0);
|
||||
|
||||
SRpcEpSet* pEpSet = &pSql->epSet;
|
||||
|
|
@ -158,8 +158,8 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
|
|||
SSqlRes *pRes = &pSql->res;
|
||||
|
||||
if (code == 0) {
|
||||
SCMHeartBeatRsp *pRsp = (SCMHeartBeatRsp *)pRes->pRsp;
|
||||
SRpcEpSet * epSet = &pRsp->epSet;
|
||||
SHeartBeatRsp *pRsp = (SHeartBeatRsp *)pRes->pRsp;
|
||||
SRpcEpSet * epSet = &pRsp->epSet;
|
||||
if (epSet->numOfEps > 0) {
|
||||
tscEpSetHtons(epSet);
|
||||
tscUpdateMgmtEpSet(epSet);
|
||||
|
|
@ -566,7 +566,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
|
|||
STableMeta * pTableMeta = pTableMetaInfo->pTableMeta;
|
||||
if (UTIL_TABLE_IS_NORMAL_TABLE(pTableMetaInfo) || pTableMetaInfo->pVgroupTables == NULL) {
|
||||
|
||||
SCMVgroupInfo* pVgroupInfo = NULL;
|
||||
SVgroupInfo* pVgroupInfo = NULL;
|
||||
if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
|
||||
int32_t index = pTableMetaInfo->vgroupIndex;
|
||||
assert(index >= 0);
|
||||
|
|
@ -907,10 +907,10 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildCreateDbMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMCreateDbMsg);
|
||||
pCmd->payloadLen = sizeof(SCreateDbMsg);
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_CREATE_DB;
|
||||
|
||||
SCMCreateDbMsg *pCreateDbMsg = (SCMCreateDbMsg*)pCmd->payload;
|
||||
SCreateDbMsg *pCreateDbMsg = (SCreateDbMsg *)pCmd->payload;
|
||||
|
||||
assert(pCmd->numOfClause == 1);
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
|
|
@ -921,13 +921,13 @@ int32_t tscBuildCreateDbMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildCreateDnodeMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMCreateDnodeMsg);
|
||||
pCmd->payloadLen = sizeof(SCreateDnodeMsg);
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
tscError("%p failed to malloc for query msg", pSql);
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMCreateDnodeMsg *pCreate = (SCMCreateDnodeMsg *)pCmd->payload;
|
||||
SCreateDnodeMsg *pCreate = (SCreateDnodeMsg *)pCmd->payload;
|
||||
strncpy(pCreate->ep, pInfo->pDCLInfo->a[0].z, pInfo->pDCLInfo->a[0].n);
|
||||
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_CREATE_DNODE;
|
||||
|
|
@ -982,14 +982,14 @@ int32_t tscBuildAcctMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildUserMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMCreateUserMsg);
|
||||
pCmd->payloadLen = sizeof(SCreateUserMsg);
|
||||
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
tscError("%p failed to malloc for query msg", pSql);
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMCreateUserMsg *pAlterMsg = (SCMCreateUserMsg*)pCmd->payload;
|
||||
SCreateUserMsg *pAlterMsg = (SCreateUserMsg *)pCmd->payload;
|
||||
|
||||
SUserInfo *pUser = &pInfo->pDCLInfo->user;
|
||||
strncpy(pAlterMsg->user, pUser->user.z, pUser->user.n);
|
||||
|
|
@ -1014,21 +1014,21 @@ int32_t tscBuildUserMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildCfgDnodeMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMCfgDnodeMsg);
|
||||
pCmd->payloadLen = sizeof(SCfgDnodeMsg);
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_CONFIG_DNODE;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t tscBuildDropDbMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMDropDbMsg);
|
||||
pCmd->payloadLen = sizeof(SDropDbMsg);
|
||||
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
tscError("%p failed to malloc for query msg", pSql);
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMDropDbMsg *pDropDbMsg = (SCMDropDbMsg*)pCmd->payload;
|
||||
SDropDbMsg *pDropDbMsg = (SDropDbMsg*)pCmd->payload;
|
||||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
tstrncpy(pDropDbMsg->db, pTableMetaInfo->name, sizeof(pDropDbMsg->db));
|
||||
|
|
@ -1058,13 +1058,13 @@ int32_t tscBuildDropTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildDropDnodeMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMDropDnodeMsg);
|
||||
pCmd->payloadLen = sizeof(SDropDnodeMsg);
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
tscError("%p failed to malloc for query msg", pSql);
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMDropDnodeMsg *pDrop = (SCMDropDnodeMsg *)pCmd->payload;
|
||||
SDropDnodeMsg * pDrop = (SDropDnodeMsg *)pCmd->payload;
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
tstrncpy(pDrop->ep, pTableMetaInfo->name, sizeof(pDrop->ep));
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_DROP_DNODE;
|
||||
|
|
@ -1074,7 +1074,7 @@ int32_t tscBuildDropDnodeMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildDropUserMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMDropUserMsg);
|
||||
pCmd->payloadLen = sizeof(SDropUserMsg);
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_DROP_USER;
|
||||
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
|
|
@ -1082,7 +1082,7 @@ int32_t tscBuildDropUserMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMDropUserMsg *pDropMsg = (SCMDropUserMsg*)pCmd->payload;
|
||||
SDropUserMsg * pDropMsg = (SDropUserMsg *)pCmd->payload;
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
tstrncpy(pDropMsg->user, pTableMetaInfo->name, sizeof(pDropMsg->user));
|
||||
|
||||
|
|
@ -1091,7 +1091,7 @@ int32_t tscBuildDropUserMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildDropAcctMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMDropUserMsg);
|
||||
pCmd->payloadLen = sizeof(SDropUserMsg);
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_DROP_ACCT;
|
||||
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
|
|
@ -1099,7 +1099,7 @@ int32_t tscBuildDropAcctMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMDropUserMsg *pDropMsg = (SCMDropUserMsg*)pCmd->payload;
|
||||
SDropUserMsg * pDropMsg = (SDropUserMsg *)pCmd->payload;
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
tstrncpy(pDropMsg->user, pTableMetaInfo->name, sizeof(pDropMsg->user));
|
||||
|
||||
|
|
@ -1108,14 +1108,14 @@ int32_t tscBuildDropAcctMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int32_t tscBuildUseDbMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMUseDbMsg);
|
||||
pCmd->payloadLen = sizeof(SUseDbMsg);
|
||||
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
tscError("%p failed to malloc for query msg", pSql);
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMUseDbMsg *pUseDbMsg = (SCMUseDbMsg*)pCmd->payload;
|
||||
SUseDbMsg *pUseDbMsg = (SUseDbMsg *)pCmd->payload;
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
strcpy(pUseDbMsg->db, pTableMetaInfo->name);
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_USE_DB;
|
||||
|
|
@ -1127,14 +1127,14 @@ int32_t tscBuildShowMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
STscObj *pObj = pSql->pTscObj;
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_SHOW;
|
||||
pCmd->payloadLen = sizeof(SCMShowMsg) + 100;
|
||||
pCmd->payloadLen = sizeof(SShowMsg) + 100;
|
||||
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
|
||||
tscError("%p failed to malloc for query msg", pSql);
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMShowMsg *pShowMsg = (SCMShowMsg*)pCmd->payload;
|
||||
SShowMsg *pShowMsg = (SShowMsg *)pCmd->payload;
|
||||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
size_t nameLen = strlen(pTableMetaInfo->name);
|
||||
|
|
@ -1161,13 +1161,13 @@ int32_t tscBuildShowMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
pShowMsg->payloadLen = htons(pEpAddr->n);
|
||||
}
|
||||
|
||||
pCmd->payloadLen = sizeof(SCMShowMsg) + pShowMsg->payloadLen;
|
||||
pCmd->payloadLen = sizeof(SShowMsg) + pShowMsg->payloadLen;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t tscBuildKillMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMKillQueryMsg);
|
||||
pCmd->payloadLen = sizeof(SKillQueryMsg);
|
||||
|
||||
switch (pCmd->command) {
|
||||
case TSDB_SQL_KILL_QUERY:
|
||||
|
|
@ -1279,8 +1279,7 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int tscEstimateAlterTableMsgLength(SSqlCmd *pCmd) {
|
||||
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
|
||||
return minMsgSize() + sizeof(SCMAlterTableMsg) + sizeof(SSchema) * tscNumOfFields(pQueryInfo) +
|
||||
TSDB_EXTRA_PAYLOAD_SIZE;
|
||||
return minMsgSize() + sizeof(SAlterTableMsg) + sizeof(SSchema) * tscNumOfFields(pQueryInfo) + TSDB_EXTRA_PAYLOAD_SIZE;
|
||||
}
|
||||
|
||||
int tscBuildAlterTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
|
|
@ -1299,7 +1298,7 @@ int tscBuildAlterTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMAlterTableMsg *pAlterTableMsg = (SCMAlterTableMsg *)pCmd->payload;
|
||||
SAlterTableMsg *pAlterTableMsg = (SAlterTableMsg *)pCmd->payload;
|
||||
tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pAlterTableMsg->db);
|
||||
|
||||
strcpy(pAlterTableMsg->tableId, pTableMetaInfo->name);
|
||||
|
|
@ -1348,10 +1347,10 @@ int tscBuildUpdateTagMsg(SSqlObj* pSql, SSqlInfo *pInfo) {
|
|||
|
||||
int tscAlterDbMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
pCmd->payloadLen = sizeof(SCMAlterDbMsg);
|
||||
pCmd->payloadLen = sizeof(SAlterDbMsg);
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_ALTER_DB;
|
||||
|
||||
SCMAlterDbMsg *pAlterDbMsg = (SCMAlterDbMsg*)pCmd->payload;
|
||||
SAlterDbMsg *pAlterDbMsg = (SAlterDbMsg* )pCmd->payload;
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
tstrncpy(pAlterDbMsg->db, pTableMetaInfo->name, sizeof(pAlterDbMsg->db));
|
||||
|
||||
|
|
@ -1505,11 +1504,11 @@ int tscBuildTableMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
|
||||
|
||||
SCMTableInfoMsg* pInfoMsg = (SCMTableInfoMsg *)pCmd->payload;
|
||||
STableInfoMsg *pInfoMsg = (STableInfoMsg *)pCmd->payload;
|
||||
strcpy(pInfoMsg->tableId, pTableMetaInfo->name);
|
||||
pInfoMsg->createFlag = htons(pSql->cmd.autoCreated ? 1 : 0);
|
||||
|
||||
char* pMsg = (char*)pInfoMsg + sizeof(SCMTableInfoMsg);
|
||||
char *pMsg = (char *)pInfoMsg + sizeof(STableInfoMsg);
|
||||
|
||||
size_t len = htonl(pCmd->tagData.dataLen);
|
||||
if (pSql->cmd.autoCreated) {
|
||||
|
|
@ -1528,7 +1527,7 @@ int tscBuildTableMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
/**
|
||||
* multi table meta req pkg format:
|
||||
* | SMgmtHead | SCMMultiTableInfoMsg | tableId0 | tableId1 | tableId2 | ......
|
||||
* | SMgmtHead | SMultiTableInfoMsg | tableId0 | tableId1 | tableId2 | ......
|
||||
* no used 4B
|
||||
**/
|
||||
int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||
|
|
@ -1546,7 +1545,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
SMgmtHead *pMgmt = (SMgmtHead *)(pCmd->payload + tsRpcHeadSize);
|
||||
memset(pMgmt->db, 0, TSDB_TABLE_FNAME_LEN); // server don't need the db
|
||||
|
||||
SCMMultiTableInfoMsg *pInfoMsg = (SCMMultiTableInfoMsg *)(pCmd->payload + tsRpcHeadSize + sizeof(SMgmtHead));
|
||||
SMultiTableInfoMsg *pInfoMsg = (SMultiTableInfoMsg *)(pCmd->payload + tsRpcHeadSize + sizeof(SMgmtHead));
|
||||
pInfoMsg->numOfTables = htonl((int32_t)pCmd->count);
|
||||
|
||||
if (pCmd->payloadLen > 0) {
|
||||
|
|
@ -1555,7 +1554,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
|
||||
taosTFree(tmpData);
|
||||
|
||||
pCmd->payloadLen += sizeof(SMgmtHead) + sizeof(SCMMultiTableInfoMsg);
|
||||
pCmd->payloadLen += sizeof(SMgmtHead) + sizeof(SMultiTableInfoMsg);
|
||||
pCmd->msgType = TSDB_MSG_TYPE_CM_TABLES_META;
|
||||
|
||||
assert(pCmd->payloadLen + minMsgSize() <= pCmd->allocSize);
|
||||
|
|
@ -1601,11 +1600,11 @@ int tscBuildSTableVgroupMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
char* pMsg = pCmd->payload;
|
||||
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
|
||||
|
||||
SCMSTableVgroupMsg *pStableVgroupMsg = (SCMSTableVgroupMsg *) pMsg;
|
||||
SSTableVgroupMsg *pStableVgroupMsg = (SSTableVgroupMsg *)pMsg;
|
||||
pStableVgroupMsg->numOfTables = htonl(pQueryInfo->numOfTables);
|
||||
pMsg += sizeof(SCMSTableVgroupMsg);
|
||||
pMsg += sizeof(SSTableVgroupMsg);
|
||||
|
||||
for(int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
|
||||
for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, i);
|
||||
size_t size = sizeof(pTableMetaInfo->name);
|
||||
tstrncpy(pMsg, pTableMetaInfo->name, size);
|
||||
|
|
@ -1638,14 +1637,14 @@ int tscBuildHeartBeatMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
|||
numOfStreams++;
|
||||
}
|
||||
|
||||
int size = numOfQueries * sizeof(SQueryDesc) + numOfStreams * sizeof(SStreamDesc) + sizeof(SCMHeartBeatMsg) + 100;
|
||||
int size = numOfQueries * sizeof(SQueryDesc) + numOfStreams * sizeof(SStreamDesc) + sizeof(SHeartBeatMsg) + 100;
|
||||
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, size)) {
|
||||
pthread_mutex_unlock(&pObj->mutex);
|
||||
tscError("%p failed to malloc for heartbeat msg", pSql);
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMHeartBeatMsg *pHeartbeat = (SCMHeartBeatMsg *)pCmd->payload;
|
||||
SHeartBeatMsg *pHeartbeat = (SHeartBeatMsg *)pCmd->payload;
|
||||
pHeartbeat->numOfQueries = numOfQueries;
|
||||
pHeartbeat->numOfStreams = numOfStreams;
|
||||
|
||||
|
|
@ -1734,7 +1733,7 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
|
|||
|
||||
/**
|
||||
* multi table meta rsp pkg format:
|
||||
* | STaosRsp | ieType | SCMMultiTableInfoMsg | SMeterMeta0 | SSchema0 | SMeterMeta1 | SSchema1 | SMeterMeta2 | SSchema2
|
||||
* | STaosRsp | ieType | SMultiTableInfoMsg | SMeterMeta0 | SSchema0 | SMeterMeta1 | SSchema1 | SMeterMeta2 | SSchema2
|
||||
* |...... 1B 1B 4B
|
||||
**/
|
||||
int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
|
||||
|
|
@ -1751,9 +1750,9 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
|
|||
|
||||
rsp++;
|
||||
|
||||
SCMMultiTableInfoMsg *pInfo = (SCMMultiTableInfoMsg *)rsp;
|
||||
SMultiTableInfoMsg *pInfo = (SMultiTableInfoMsg *)rsp;
|
||||
totalNum = htonl(pInfo->numOfTables);
|
||||
rsp += sizeof(SCMMultiTableInfoMsg);
|
||||
rsp += sizeof(SMultiTableInfoMsg);
|
||||
|
||||
for (i = 0; i < totalNum; i++) {
|
||||
SMultiTableMeta *pMultiMeta = (SMultiTableMeta *)rsp;
|
||||
|
|
@ -1845,9 +1844,9 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
|
|||
SSqlRes* pRes = &pSql->res;
|
||||
|
||||
// NOTE: the order of several table must be preserved.
|
||||
SCMSTableVgroupRspMsg *pStableVgroup = (SCMSTableVgroupRspMsg *)pRes->pRsp;
|
||||
SSTableVgroupRspMsg *pStableVgroup = (SSTableVgroupRspMsg *)pRes->pRsp;
|
||||
pStableVgroup->numOfTables = htonl(pStableVgroup->numOfTables);
|
||||
char* pMsg = pRes->pRsp + sizeof(SCMSTableVgroupRspMsg);
|
||||
char *pMsg = pRes->pRsp + sizeof(SSTableVgroupRspMsg);
|
||||
|
||||
// master sqlObj locates in param
|
||||
SSqlObj* parent = pSql->param;
|
||||
|
|
@ -1860,18 +1859,18 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
|
|||
SVgroupsMsg * pVgroupMsg = (SVgroupsMsg *) pMsg;
|
||||
pVgroupMsg->numOfVgroups = htonl(pVgroupMsg->numOfVgroups);
|
||||
|
||||
size_t size = sizeof(SCMVgroupMsg) * pVgroupMsg->numOfVgroups + sizeof(SVgroupsMsg);
|
||||
size_t size = sizeof(SVgroupMsg) * pVgroupMsg->numOfVgroups + sizeof(SVgroupsMsg);
|
||||
|
||||
size_t vgroupsz = sizeof(SCMVgroupInfo) * pVgroupMsg->numOfVgroups + sizeof(SVgroupsInfo);
|
||||
size_t vgroupsz = sizeof(SVgroupInfo) * pVgroupMsg->numOfVgroups + sizeof(SVgroupsInfo);
|
||||
pInfo->vgroupList = calloc(1, vgroupsz);
|
||||
assert(pInfo->vgroupList != NULL);
|
||||
|
||||
pInfo->vgroupList->numOfVgroups = pVgroupMsg->numOfVgroups;
|
||||
for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) {
|
||||
//just init, no need to lock
|
||||
SCMVgroupInfo *pVgroups = &pInfo->vgroupList->vgroups[j];
|
||||
SVgroupInfo *pVgroups = &pInfo->vgroupList->vgroups[j];
|
||||
|
||||
SCMVgroupMsg *vmsg = &pVgroupMsg->vgroups[j];
|
||||
SVgroupMsg *vmsg = &pVgroupMsg->vgroups[j];
|
||||
pVgroups->vgId = htonl(vmsg->vgId);
|
||||
pVgroups->numOfEps = vmsg->numOfEps;
|
||||
|
||||
|
|
@ -1893,10 +1892,10 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
|
|||
* current process do not use the cache at all
|
||||
*/
|
||||
int tscProcessShowRsp(SSqlObj *pSql) {
|
||||
STableMetaMsg * pMetaMsg;
|
||||
SCMShowRsp *pShow;
|
||||
SSchema * pSchema;
|
||||
char key[20];
|
||||
STableMetaMsg *pMetaMsg;
|
||||
SShowRsp * pShow;
|
||||
SSchema * pSchema;
|
||||
char key[20];
|
||||
|
||||
SSqlRes *pRes = &pSql->res;
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
|
|
@ -1905,7 +1904,7 @@ int tscProcessShowRsp(SSqlObj *pSql) {
|
|||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
|
||||
|
||||
pShow = (SCMShowRsp *)pRes->pRsp;
|
||||
pShow = (SShowRsp *)pRes->pRsp;
|
||||
pShow->qhandle = htobe64(pShow->qhandle);
|
||||
pRes->qhandle = pShow->qhandle;
|
||||
|
||||
|
|
|
|||
|
|
@ -489,7 +489,7 @@ void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArr
|
|||
SVgroupTableInfo info = {{0}};
|
||||
for (int32_t m = 0; m < pvg->numOfVgroups; ++m) {
|
||||
if (tt->vgId == pvg->vgroups[m].vgId) {
|
||||
tscSCMVgroupInfoCopy(&info.vgInfo, &pvg->vgroups[m]);
|
||||
tscSVgroupInfoCopy(&info.vgInfo, &pvg->vgroups[m]);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
|
@ -1645,8 +1645,8 @@ static int32_t tscReissueSubquery(SRetrieveSupport *trsupport, SSqlObj *pSql, in
|
|||
SSqlObj *pParentSql = trsupport->pParentSql;
|
||||
int32_t subqueryIndex = trsupport->subqueryIndex;
|
||||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, 0, 0);
|
||||
SCMVgroupInfo* pVgroup = &pTableMetaInfo->vgroupList->vgroups[0];
|
||||
STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, 0, 0);
|
||||
SVgroupInfo* pVgroup = &pTableMetaInfo->vgroupList->vgroups[0];
|
||||
|
||||
tExtMemBufferClear(trsupport->pExtMemBuffer[subqueryIndex]);
|
||||
|
||||
|
|
@ -1848,7 +1848,7 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
|
|||
assert(pState->numOfRemain <= pState->numOfSub && pState->numOfRemain >= 0);
|
||||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, 0, 0);
|
||||
SCMVgroupInfo *pVgroup = &pTableMetaInfo->vgroupList->vgroups[0];
|
||||
SVgroupInfo *pVgroup = &pTableMetaInfo->vgroupList->vgroups[0];
|
||||
|
||||
if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
|
||||
trsupport->numOfRetry = MAX_NUM_OF_SUBQUERY_RETRY;
|
||||
|
|
@ -1959,7 +1959,7 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
|
|||
assert(pSql->cmd.numOfClause == 1 && pQueryInfo->numOfTables == 1);
|
||||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, 0, 0);
|
||||
SCMVgroupInfo* pVgroup = &pTableMetaInfo->vgroupList->vgroups[trsupport->subqueryIndex];
|
||||
SVgroupInfo* pVgroup = &pTableMetaInfo->vgroupList->vgroups[trsupport->subqueryIndex];
|
||||
|
||||
// stable query killed or other subquery failed, all query stopped
|
||||
if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
|
||||
|
|
|
|||
|
|
@ -2424,7 +2424,7 @@ SVgroupsInfo* tscVgroupInfoClone(SVgroupsInfo *vgroupList) {
|
|||
return NULL;
|
||||
}
|
||||
|
||||
size_t size = sizeof(SVgroupsInfo) + sizeof(SCMVgroupInfo) * vgroupList->numOfVgroups;
|
||||
size_t size = sizeof(SVgroupsInfo) + sizeof(SVgroupInfo) * vgroupList->numOfVgroups;
|
||||
SVgroupsInfo* pNew = calloc(1, size);
|
||||
if (pNew == NULL) {
|
||||
return NULL;
|
||||
|
|
@ -2433,9 +2433,9 @@ SVgroupsInfo* tscVgroupInfoClone(SVgroupsInfo *vgroupList) {
|
|||
pNew->numOfVgroups = vgroupList->numOfVgroups;
|
||||
|
||||
for(int32_t i = 0; i < vgroupList->numOfVgroups; ++i) {
|
||||
SCMVgroupInfo* pNewVInfo = &pNew->vgroups[i];
|
||||
SVgroupInfo* pNewVInfo = &pNew->vgroups[i];
|
||||
|
||||
SCMVgroupInfo* pvInfo = &vgroupList->vgroups[i];
|
||||
SVgroupInfo* pvInfo = &vgroupList->vgroups[i];
|
||||
pNewVInfo->vgId = pvInfo->vgId;
|
||||
pNewVInfo->numOfEps = pvInfo->numOfEps;
|
||||
|
||||
|
|
@ -2454,7 +2454,7 @@ void* tscVgroupInfoClear(SVgroupsInfo *vgroupList) {
|
|||
}
|
||||
|
||||
for(int32_t i = 0; i < vgroupList->numOfVgroups; ++i) {
|
||||
SCMVgroupInfo* pVgroupInfo = &vgroupList->vgroups[i];
|
||||
SVgroupInfo* pVgroupInfo = &vgroupList->vgroups[i];
|
||||
|
||||
for(int32_t j = 0; j < pVgroupInfo->numOfEps; ++j) {
|
||||
taosTFree(pVgroupInfo->epAddr[j].fqdn);
|
||||
|
|
@ -2465,7 +2465,7 @@ void* tscVgroupInfoClear(SVgroupsInfo *vgroupList) {
|
|||
return NULL;
|
||||
}
|
||||
|
||||
void tscSCMVgroupInfoCopy(SCMVgroupInfo* dst, const SCMVgroupInfo* src) {
|
||||
void tscSVgroupInfoCopy(SVgroupInfo* dst, const SVgroupInfo* src) {
|
||||
dst->vgId = src->vgId;
|
||||
dst->numOfEps = src->numOfEps;
|
||||
for(int32_t i = 0; i < dst->numOfEps; ++i) {
|
||||
|
|
|
|||
|
|
@ -352,7 +352,7 @@ static void dnodeCloseVnodes() {
|
|||
}
|
||||
|
||||
static void* dnodeParseVnodeMsg(SRpcMsg *rpcMsg) {
|
||||
SMDCreateVnodeMsg *pCreate = rpcMsg->pCont;
|
||||
SCreateVnodeMsg *pCreate = rpcMsg->pCont;
|
||||
pCreate->cfg.vgId = htonl(pCreate->cfg.vgId);
|
||||
pCreate->cfg.cfgVersion = htonl(pCreate->cfg.cfgVersion);
|
||||
pCreate->cfg.maxTables = htonl(pCreate->cfg.maxTables);
|
||||
|
|
@ -375,7 +375,7 @@ static void* dnodeParseVnodeMsg(SRpcMsg *rpcMsg) {
|
|||
}
|
||||
|
||||
static int32_t dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
|
||||
SMDCreateVnodeMsg *pCreate = dnodeParseVnodeMsg(rpcMsg);
|
||||
SCreateVnodeMsg *pCreate = dnodeParseVnodeMsg(rpcMsg);
|
||||
|
||||
void *pVnode = vnodeAcquire(pCreate->cfg.vgId);
|
||||
if (pVnode != NULL) {
|
||||
|
|
@ -389,7 +389,7 @@ static int32_t dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
|
|||
}
|
||||
|
||||
static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
|
||||
SMDAlterVnodeMsg *pAlter = dnodeParseVnodeMsg(rpcMsg);
|
||||
SAlterVnodeMsg *pAlter = dnodeParseVnodeMsg(rpcMsg);
|
||||
|
||||
void *pVnode = vnodeAcquire(pAlter->cfg.vgId);
|
||||
if (pVnode != NULL) {
|
||||
|
|
@ -404,14 +404,14 @@ static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
|
|||
}
|
||||
|
||||
static int32_t dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) {
|
||||
SMDDropVnodeMsg *pDrop = rpcMsg->pCont;
|
||||
SDropVnodeMsg *pDrop = rpcMsg->pCont;
|
||||
pDrop->vgId = htonl(pDrop->vgId);
|
||||
|
||||
return vnodeDrop(pDrop->vgId);
|
||||
}
|
||||
|
||||
static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) {
|
||||
// SMDAlterStreamMsg *pStream = pCont;
|
||||
// SAlterStreamMsg *pStream = pCont;
|
||||
// pStream->uid = htobe64(pStream->uid);
|
||||
// pStream->stime = htobe64(pStream->stime);
|
||||
// pStream->vnode = htonl(pStream->vnode);
|
||||
|
|
@ -424,12 +424,12 @@ static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
|
||||
SMDCfgDnodeMsg *pCfg = pMsg->pCont;
|
||||
SCfgDnodeMsg *pCfg = pMsg->pCont;
|
||||
return taosCfgDynamicOptions(pCfg->config);
|
||||
}
|
||||
|
||||
static int32_t dnodeProcessCreateMnodeMsg(SRpcMsg *pMsg) {
|
||||
SMDCreateMnodeMsg *pCfg = pMsg->pCont;
|
||||
SCreateMnodeMsg *pCfg = pMsg->pCont;
|
||||
pCfg->dnodeId = htonl(pCfg->dnodeId);
|
||||
if (pCfg->dnodeId != dnodeGetDnodeId()) {
|
||||
dError("dnodeId:%d, in create mnode msg is not equal with saved dnodeId:%d", pCfg->dnodeId, dnodeGetDnodeId());
|
||||
|
|
@ -459,7 +459,7 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
|
|||
return;
|
||||
}
|
||||
|
||||
SDMStatusRsp *pStatusRsp = pMsg->pCont;
|
||||
SStatusRsp *pStatusRsp = pMsg->pCont;
|
||||
SMnodeInfos *minfos = &pStatusRsp->mnodes;
|
||||
dnodeUpdateMInfos(minfos);
|
||||
|
||||
|
|
@ -471,7 +471,7 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
|
|||
|
||||
vnodeSetAccess(pStatusRsp->vgAccess, pCfg->numOfVnodes);
|
||||
|
||||
SDnodeEps *pEps = (SDnodeEps *)((char *)pStatusRsp->vgAccess + pCfg->numOfVnodes * sizeof(SDMVgroupAccess));
|
||||
SDnodeEps *pEps = (SDnodeEps *)((char *)pStatusRsp->vgAccess + pCfg->numOfVnodes * sizeof(SVgroupAccess));
|
||||
dnodeUpdateEps(pEps);
|
||||
|
||||
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
|
||||
|
|
@ -489,8 +489,8 @@ static void dnodeSendStatusMsg(void *handle, void *tmrId) {
|
|||
return;
|
||||
}
|
||||
|
||||
int32_t contLen = sizeof(SDMStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
|
||||
SDMStatusMsg *pStatus = rpcMallocCont(contLen);
|
||||
int32_t contLen = sizeof(SStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
|
||||
SStatusMsg *pStatus = rpcMallocCont(contLen);
|
||||
if (pStatus == NULL) {
|
||||
taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
|
||||
dError("failed to malloc status message");
|
||||
|
|
@ -523,7 +523,7 @@ static void dnodeSendStatusMsg(void *handle, void *tmrId) {
|
|||
tstrncpy(pStatus->clusterCfg.charset, tsCharset, TSDB_LOCALE_LEN);
|
||||
|
||||
vnodeBuildStatusMsg(pStatus);
|
||||
contLen = sizeof(SDMStatusMsg) + pStatus->openVnodes * sizeof(SVnodeLoad);
|
||||
contLen = sizeof(SStatusMsg) + pStatus->openVnodes * sizeof(SVnodeLoad);
|
||||
pStatus->openVnodes = htons(pStatus->openVnodes);
|
||||
|
||||
SRpcMsg rpcMsg = {
|
||||
|
|
|
|||
|
|
@ -137,7 +137,7 @@ int32_t dnodeInitClient() {
|
|||
return -1;
|
||||
}
|
||||
|
||||
dInfo("initialized inter-dnodes rpc client is initialized");
|
||||
dInfo("dnode inter-dnodes rpc client is initialized");
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -145,7 +145,7 @@ void dnodeCleanupClient() {
|
|||
if (tsDnodeClientRpc) {
|
||||
rpcClose(tsDnodeClientRpc);
|
||||
tsDnodeClientRpc = NULL;
|
||||
dInfo("inter-dnodes rpc client is closed");
|
||||
dInfo("dnode inter-dnodes rpc client is closed");
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -146,12 +146,12 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
|
|||
int code = mnodeRetriveAuth(user, spi, encrypt, secret, ckey);
|
||||
if (code != TSDB_CODE_APP_NOT_READY) return code;
|
||||
|
||||
SDMAuthMsg *pMsg = rpcMallocCont(sizeof(SDMAuthMsg));
|
||||
SAuthMsg *pMsg = rpcMallocCont(sizeof(SAuthMsg));
|
||||
tstrncpy(pMsg->user, user, sizeof(pMsg->user));
|
||||
|
||||
SRpcMsg rpcMsg = {0};
|
||||
rpcMsg.pCont = pMsg;
|
||||
rpcMsg.contLen = sizeof(SDMAuthMsg);
|
||||
rpcMsg.contLen = sizeof(SAuthMsg);
|
||||
rpcMsg.msgType = TSDB_MSG_TYPE_DM_AUTH;
|
||||
|
||||
dDebug("user:%s, send auth msg to mnodes", user);
|
||||
|
|
@ -161,7 +161,7 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
|
|||
if (rpcRsp.code != 0) {
|
||||
dError("user:%s, auth msg received from mnodes, error:%s", user, tstrerror(rpcRsp.code));
|
||||
} else {
|
||||
SDMAuthRsp *pRsp = rpcRsp.pCont;
|
||||
SAuthRsp *pRsp = rpcRsp.pCont;
|
||||
dDebug("user:%s, auth msg received from mnodes", user);
|
||||
memcpy(secret, pRsp->secret, TSDB_KEY_LEN);
|
||||
memcpy(ckey, pRsp->ckey, TSDB_KEY_LEN);
|
||||
|
|
@ -176,8 +176,8 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
|
|||
void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid) {
|
||||
dDebug("vgId:%d, tid:%d send config table msg to mnode", vgId, tid);
|
||||
|
||||
int32_t contLen = sizeof(SDMConfigTableMsg);
|
||||
SDMConfigTableMsg *pMsg = rpcMallocCont(contLen);
|
||||
int32_t contLen = sizeof(SConfigTableMsg);
|
||||
SConfigTableMsg *pMsg = rpcMallocCont(contLen);
|
||||
|
||||
pMsg->dnodeId = htonl(dnodeGetDnodeId());
|
||||
pMsg->vgId = htonl(vgId);
|
||||
|
|
|
|||
|
|
@ -291,7 +291,7 @@ typedef struct {
|
|||
SSchema schema[];
|
||||
// tagVal is padded after schema
|
||||
// char tagVal[];
|
||||
} SCMAlterTableMsg;
|
||||
} SAlterTableMsg;
|
||||
|
||||
typedef struct {
|
||||
SMsgHead head;
|
||||
|
|
@ -348,14 +348,14 @@ typedef struct {
|
|||
|
||||
typedef struct {
|
||||
char user[TSDB_USER_LEN];
|
||||
} SCMDropUserMsg, SCMDropAcctMsg;
|
||||
} SDropUserMsg, SDropAcctMsg;
|
||||
|
||||
typedef struct {
|
||||
char user[TSDB_USER_LEN];
|
||||
char pass[TSDB_KEY_LEN];
|
||||
int8_t privilege;
|
||||
int8_t flag;
|
||||
} SCMCreateUserMsg, SCMAlterUserMsg;
|
||||
} SCreateUserMsg, SAlterUserMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t contLen;
|
||||
|
|
@ -370,11 +370,11 @@ typedef struct {
|
|||
int32_t vgId;
|
||||
uint64_t uid;
|
||||
char tableId[TSDB_TABLE_FNAME_LEN];
|
||||
} SMDDropSTableMsg;
|
||||
} SDropSTableMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t vgId;
|
||||
} SMDDropVnodeMsg;
|
||||
} SDropVnodeMsg;
|
||||
|
||||
typedef struct SColIndex {
|
||||
int16_t colId; // column id
|
||||
|
|
@ -540,12 +540,12 @@ typedef struct {
|
|||
int8_t replications;
|
||||
int8_t quorum;
|
||||
int8_t ignoreExist;
|
||||
} SCMCreateDbMsg, SCMAlterDbMsg;
|
||||
} SCreateDbMsg, SAlterDbMsg;
|
||||
|
||||
typedef struct {
|
||||
char db[TSDB_TABLE_FNAME_LEN];
|
||||
uint8_t ignoreNotExists;
|
||||
} SCMDropDbMsg, SCMUseDbMsg;
|
||||
} SDropDbMsg, SUseDbMsg;
|
||||
|
||||
// IMPORTANT: sizeof(SVnodeStatisticInfo) should not exceed
|
||||
// TSDB_FILE_HEADER_LEN/4 - TSDB_FILE_HEADER_VERSION_SIZE
|
||||
|
|
@ -560,7 +560,7 @@ typedef struct {
|
|||
typedef struct {
|
||||
int32_t vgId;
|
||||
int8_t accessState;
|
||||
} SDMVgroupAccess;
|
||||
} SVgroupAccess;
|
||||
|
||||
typedef struct {
|
||||
int32_t dnodeId;
|
||||
|
|
@ -622,13 +622,13 @@ typedef struct {
|
|||
uint8_t reserve2[15];
|
||||
SClusterCfg clusterCfg;
|
||||
SVnodeLoad load[];
|
||||
} SDMStatusMsg;
|
||||
} SStatusMsg;
|
||||
|
||||
typedef struct {
|
||||
SMnodeInfos mnodes;
|
||||
SDnodeCfg dnodeCfg;
|
||||
SDMVgroupAccess vgAccess[];
|
||||
} SDMStatusRsp;
|
||||
SMnodeInfos mnodes;
|
||||
SDnodeCfg dnodeCfg;
|
||||
SVgroupAccess vgAccess[];
|
||||
} SStatusRsp;
|
||||
|
||||
typedef struct {
|
||||
uint32_t vgId;
|
||||
|
|
@ -651,54 +651,54 @@ typedef struct {
|
|||
int8_t wals;
|
||||
int8_t quorum;
|
||||
int8_t reserved[16];
|
||||
} SMDVnodeCfg;
|
||||
} SVnodeCfg;
|
||||
|
||||
typedef struct {
|
||||
int32_t nodeId;
|
||||
char nodeEp[TSDB_EP_LEN];
|
||||
} SMDVnodeDesc;
|
||||
} SVnodeDesc;
|
||||
|
||||
typedef struct {
|
||||
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
|
||||
SMDVnodeCfg cfg;
|
||||
SMDVnodeDesc nodes[TSDB_MAX_REPLICA];
|
||||
} SMDCreateVnodeMsg, SMDAlterVnodeMsg;
|
||||
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
|
||||
SVnodeCfg cfg;
|
||||
SVnodeDesc nodes[TSDB_MAX_REPLICA];
|
||||
} SCreateVnodeMsg, SAlterVnodeMsg;
|
||||
|
||||
typedef struct {
|
||||
char tableId[TSDB_TABLE_FNAME_LEN];
|
||||
int16_t createFlag;
|
||||
char tags[];
|
||||
} SCMTableInfoMsg;
|
||||
} STableInfoMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t numOfTables;
|
||||
char tableIds[];
|
||||
} SCMMultiTableInfoMsg;
|
||||
} SMultiTableInfoMsg;
|
||||
|
||||
typedef struct SCMSTableVgroupMsg {
|
||||
typedef struct SSTableVgroupMsg {
|
||||
int32_t numOfTables;
|
||||
} SCMSTableVgroupMsg, SCMSTableVgroupRspMsg;
|
||||
} SSTableVgroupMsg, SSTableVgroupRspMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t vgId;
|
||||
int8_t numOfEps;
|
||||
SEpAddr1 epAddr[TSDB_MAX_REPLICA];
|
||||
} SCMVgroupInfo;
|
||||
} SVgroupInfo;
|
||||
|
||||
typedef struct {
|
||||
int32_t vgId;
|
||||
int8_t numOfEps;
|
||||
SEpAddrMsg epAddr[TSDB_MAX_REPLICA];
|
||||
} SCMVgroupMsg;
|
||||
} SVgroupMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t numOfVgroups;
|
||||
SCMVgroupInfo vgroups[];
|
||||
SVgroupInfo vgroups[];
|
||||
} SVgroupsInfo;
|
||||
|
||||
typedef struct {
|
||||
int32_t numOfVgroups;
|
||||
SCMVgroupMsg vgroups[];
|
||||
SVgroupMsg vgroups[];
|
||||
} SVgroupsMsg;
|
||||
|
||||
typedef struct STableMetaMsg {
|
||||
|
|
@ -713,7 +713,7 @@ typedef struct STableMetaMsg {
|
|||
int16_t tversion;
|
||||
int32_t tid;
|
||||
uint64_t uid;
|
||||
SCMVgroupMsg vgroup;
|
||||
SVgroupMsg vgroup;
|
||||
SSchema schema[];
|
||||
} STableMetaMsg;
|
||||
|
||||
|
|
@ -739,38 +739,38 @@ typedef struct {
|
|||
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
|
||||
uint16_t payloadLen;
|
||||
char payload[];
|
||||
} SCMShowMsg;
|
||||
} SShowMsg;
|
||||
|
||||
typedef struct SCMShowRsp {
|
||||
typedef struct SShowRsp {
|
||||
uint64_t qhandle;
|
||||
STableMetaMsg tableMeta;
|
||||
} SCMShowRsp;
|
||||
} SShowRsp;
|
||||
|
||||
typedef struct {
|
||||
char ep[TSDB_EP_LEN]; // end point, hostname:port
|
||||
} SCMCreateDnodeMsg, SCMDropDnodeMsg;
|
||||
char ep[TSDB_EP_LEN]; // end point, hostname:port
|
||||
} SCreateDnodeMsg, SDropDnodeMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t dnodeId;
|
||||
char dnodeEp[TSDB_EP_LEN]; // end point, hostname:port
|
||||
SMnodeInfos mnodes;
|
||||
} SMDCreateMnodeMsg;
|
||||
} SCreateMnodeMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t dnodeId;
|
||||
int32_t vgId;
|
||||
int32_t tid;
|
||||
} SDMConfigTableMsg;
|
||||
} SConfigTableMsg;
|
||||
|
||||
typedef struct {
|
||||
uint32_t dnodeId;
|
||||
int32_t vgId;
|
||||
} SDMConfigVnodeMsg;
|
||||
} SConfigVnodeMsg;
|
||||
|
||||
typedef struct {
|
||||
char ep[TSDB_EP_LEN]; // end point, hostname:port
|
||||
char config[64];
|
||||
} SMDCfgDnodeMsg, SCMCfgDnodeMsg;
|
||||
} SCfgDnodeMsg;
|
||||
|
||||
typedef struct {
|
||||
char sql[TSDB_SHOW_SQL_LEN];
|
||||
|
|
@ -798,7 +798,7 @@ typedef struct {
|
|||
int32_t numOfStreams;
|
||||
char appName[TSDB_APPNAME_LEN];
|
||||
char pData[];
|
||||
} SCMHeartBeatMsg;
|
||||
} SHeartBeatMsg;
|
||||
|
||||
typedef struct {
|
||||
uint32_t queryId;
|
||||
|
|
@ -808,11 +808,11 @@ typedef struct {
|
|||
uint32_t connId;
|
||||
int8_t killConnection;
|
||||
SRpcEpSet epSet;
|
||||
} SCMHeartBeatRsp;
|
||||
} SHeartBeatRsp;
|
||||
|
||||
typedef struct {
|
||||
char queryId[TSDB_KILL_MSG_LEN + 1];
|
||||
} SCMKillQueryMsg, SCMKillStreamMsg, SCMKillConnMsg;
|
||||
} SKillQueryMsg, SKillStreamMsg, SKillConnMsg;
|
||||
|
||||
typedef struct {
|
||||
int32_t vnode;
|
||||
|
|
@ -821,7 +821,7 @@ typedef struct {
|
|||
uint64_t stime; // stream starting time
|
||||
int32_t status;
|
||||
char tableId[TSDB_TABLE_FNAME_LEN];
|
||||
} SMDAlterStreamMsg;
|
||||
} SAlterStreamMsg;
|
||||
|
||||
typedef struct {
|
||||
char user[TSDB_USER_LEN];
|
||||
|
|
@ -829,7 +829,7 @@ typedef struct {
|
|||
char encrypt;
|
||||
char secret[TSDB_KEY_LEN];
|
||||
char ckey[TSDB_KEY_LEN];
|
||||
} SDMAuthMsg, SDMAuthRsp;
|
||||
} SAuthMsg, SAuthRsp;
|
||||
|
||||
#pragma pack(pop)
|
||||
|
||||
|
|
|
|||
|
|
@ -43,10 +43,10 @@ typedef struct {
|
|||
|
||||
extern char *vnodeStatus[];
|
||||
|
||||
int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg);
|
||||
int32_t vnodeCreate(SCreateVnodeMsg *pVnodeCfg);
|
||||
int32_t vnodeDrop(int32_t vgId);
|
||||
int32_t vnodeOpen(int32_t vgId, char *rootDir);
|
||||
int32_t vnodeAlter(void *pVnode, SMDCreateVnodeMsg *pVnodeCfg);
|
||||
int32_t vnodeAlter(void *pVnode, SCreateVnodeMsg *pVnodeCfg);
|
||||
int32_t vnodeClose(int32_t vgId);
|
||||
|
||||
void* vnodeAcquire(int32_t vgId); // add refcount
|
||||
|
|
@ -60,7 +60,7 @@ int32_t vnodeCheckWrite(void *pVnode);
|
|||
int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes);
|
||||
void vnodeBuildStatusMsg(void *param);
|
||||
void vnodeConfirmForward(void *param, uint64_t version, int32_t code);
|
||||
void vnodeSetAccess(SDMVgroupAccess *pAccess, int32_t numOfVnodes);
|
||||
void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes);
|
||||
|
||||
int32_t vnodeInitResources();
|
||||
void vnodeCleanupResources();
|
||||
|
|
|
|||
|
|
@ -45,7 +45,7 @@ void mnodeCleanupProfile();
|
|||
SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port, int32_t pid, const char* app);
|
||||
SConnObj *mnodeAccquireConn(int32_t connId, char *user, uint32_t ip, uint16_t port);
|
||||
void mnodeReleaseConn(SConnObj *pConn);
|
||||
int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SCMHeartBeatMsg *pHBMsg);
|
||||
int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pHBMsg);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@
|
|||
static void * tsDbSdb = NULL;
|
||||
static int32_t tsDbUpdateSize;
|
||||
|
||||
static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate, SMnodeMsg *pMsg);
|
||||
static int32_t mnodeCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate, SMnodeMsg *pMsg);
|
||||
static int32_t mnodeDropDb(SMnodeMsg *newMsg);
|
||||
static int32_t mnodeSetDbDropping(SDbObj *pDb);
|
||||
static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
|
||||
|
|
@ -352,7 +352,7 @@ static int32_t mnodeCreateDbCb(SMnodeMsg *pMsg, int32_t code) {
|
|||
return code;
|
||||
}
|
||||
|
||||
static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate, SMnodeMsg *pMsg) {
|
||||
static int32_t mnodeCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate, SMnodeMsg *pMsg) {
|
||||
int32_t code = acctCheck(pAcct, ACCT_GRANT_DB);
|
||||
if (code != 0) return code;
|
||||
|
||||
|
|
@ -805,7 +805,7 @@ static int32_t mnodeSetDbDropping(SDbObj *pDb) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessCreateDbMsg(SMnodeMsg *pMsg) {
|
||||
SCMCreateDbMsg *pCreate = pMsg->rpcMsg.pCont;
|
||||
SCreateDbMsg *pCreate = pMsg->rpcMsg.pCont;
|
||||
pCreate->maxTables = htonl(pCreate->maxTables);
|
||||
pCreate->cacheBlockSize = htonl(pCreate->cacheBlockSize);
|
||||
pCreate->totalBlocks = htonl(pCreate->totalBlocks);
|
||||
|
|
@ -830,7 +830,7 @@ static int32_t mnodeProcessCreateDbMsg(SMnodeMsg *pMsg) {
|
|||
return code;
|
||||
}
|
||||
|
||||
static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
|
||||
static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SAlterDbMsg *pAlter) {
|
||||
SDbCfg newCfg = pDb->cfg;
|
||||
int32_t maxTables = htonl(pAlter->maxTables);
|
||||
int32_t cacheBlockSize = htonl(pAlter->cacheBlockSize);
|
||||
|
|
@ -977,7 +977,7 @@ static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t mnodeAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter, void *pMsg) {
|
||||
static int32_t mnodeAlterDb(SDbObj *pDb, SAlterDbMsg *pAlter, void *pMsg) {
|
||||
SDbCfg newCfg = mnodeGetAlterDbOption(pDb, pAlter);
|
||||
if (terrno != TSDB_CODE_SUCCESS) {
|
||||
return terrno;
|
||||
|
|
@ -1009,7 +1009,7 @@ static int32_t mnodeAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter, void *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg) {
|
||||
SCMAlterDbMsg *pAlter = pMsg->rpcMsg.pCont;
|
||||
SAlterDbMsg *pAlter = pMsg->rpcMsg.pCont;
|
||||
mDebug("db:%s, alter db msg is received from thandle:%p", pAlter->db, pMsg->rpcMsg.handle);
|
||||
|
||||
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pAlter->db);
|
||||
|
|
@ -1060,7 +1060,7 @@ static int32_t mnodeDropDb(SMnodeMsg *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessDropDbMsg(SMnodeMsg *pMsg) {
|
||||
SCMDropDbMsg *pDrop = pMsg->rpcMsg.pCont;
|
||||
SDropDbMsg *pDrop = pMsg->rpcMsg.pCont;
|
||||
mDebug("db:%s, drop db msg is received from thandle:%p", pDrop->db, pMsg->rpcMsg.handle);
|
||||
|
||||
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pDrop->db);
|
||||
|
|
|
|||
|
|
@ -314,7 +314,7 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
|
|||
return TSDB_CODE_MND_NO_RIGHTS;
|
||||
}
|
||||
|
||||
SCMCfgDnodeMsg *pCmCfgDnode = pMsg->rpcMsg.pCont;
|
||||
SCfgDnodeMsg *pCmCfgDnode = pMsg->rpcMsg.pCont;
|
||||
if (pCmCfgDnode->ep[0] == 0) {
|
||||
tstrncpy(pCmCfgDnode->ep, tsLocalEp, TSDB_EP_LEN);
|
||||
}
|
||||
|
|
@ -349,7 +349,7 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
|
|||
mnodeDecDnodeRef(pDnode);
|
||||
return code;
|
||||
} else {
|
||||
SMDCfgDnodeMsg *pMdCfgDnode = rpcMallocCont(sizeof(SMDCfgDnodeMsg));
|
||||
SCfgDnodeMsg *pMdCfgDnode = rpcMallocCont(sizeof(SCfgDnodeMsg));
|
||||
strcpy(pMdCfgDnode->ep, pCmCfgDnode->ep);
|
||||
strcpy(pMdCfgDnode->config, pCmCfgDnode->config);
|
||||
|
||||
|
|
@ -358,7 +358,7 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
|
|||
.code = 0,
|
||||
.msgType = TSDB_MSG_TYPE_MD_CONFIG_DNODE,
|
||||
.pCont = pMdCfgDnode,
|
||||
.contLen = sizeof(SMDCfgDnodeMsg)
|
||||
.contLen = sizeof(SCfgDnodeMsg)
|
||||
};
|
||||
|
||||
mInfo("dnode:%s, is configured by %s", pCmCfgDnode->ep, pMsg->pUser->user);
|
||||
|
|
@ -480,7 +480,7 @@ static void mnodeUpdateDnodeEps() {
|
|||
|
||||
static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
|
||||
SDnodeObj *pDnode = NULL;
|
||||
SDMStatusMsg *pStatus = pMsg->rpcMsg.pCont;
|
||||
SStatusMsg *pStatus = pMsg->rpcMsg.pCont;
|
||||
pStatus->dnodeId = htonl(pStatus->dnodeId);
|
||||
pStatus->moduleStatus = htonl(pStatus->moduleStatus);
|
||||
pStatus->lastReboot = htonl(pStatus->lastReboot);
|
||||
|
|
@ -538,10 +538,10 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
|
|||
|
||||
int32_t openVnodes = htons(pStatus->openVnodes);
|
||||
int32_t epsSize = mnodeGetDnodeEpsSize();
|
||||
int32_t vgAccessSize = openVnodes * sizeof(SDMVgroupAccess);
|
||||
int32_t contLen = sizeof(SDMStatusRsp) + vgAccessSize + epsSize;
|
||||
int32_t vgAccessSize = openVnodes * sizeof(SVgroupAccess);
|
||||
int32_t contLen = sizeof(SStatusRsp) + vgAccessSize + epsSize;
|
||||
|
||||
SDMStatusRsp *pRsp = rpcMallocCont(contLen);
|
||||
SStatusRsp *pRsp = rpcMallocCont(contLen);
|
||||
if (pRsp == NULL) {
|
||||
mnodeDecDnodeRef(pDnode);
|
||||
return TSDB_CODE_MND_OUT_OF_MEMORY;
|
||||
|
|
@ -551,7 +551,7 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
|
|||
pRsp->dnodeCfg.moduleStatus = htonl((int32_t)pDnode->isMgmt);
|
||||
pRsp->dnodeCfg.numOfVnodes = htonl(openVnodes);
|
||||
tstrncpy(pRsp->dnodeCfg.clusterId, mnodeGetClusterId(), TSDB_CLUSTER_ID_LEN);
|
||||
SDMVgroupAccess *pAccess = (SDMVgroupAccess *)((char *)pRsp + sizeof(SDMStatusRsp));
|
||||
SVgroupAccess *pAccess = (SVgroupAccess *)((char *)pRsp + sizeof(SStatusRsp));
|
||||
|
||||
for (int32_t j = 0; j < openVnodes; ++j) {
|
||||
SVnodeLoad *pVload = &pStatus->load[j];
|
||||
|
|
@ -602,7 +602,7 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
|
|||
|
||||
mnodeDecDnodeRef(pDnode);
|
||||
|
||||
SDnodeEps *pEps = (SDnodeEps *)((char *)pRsp + sizeof(SDMStatusRsp) + vgAccessSize);
|
||||
SDnodeEps *pEps = (SDnodeEps *)((char *)pRsp + sizeof(SStatusRsp) + vgAccessSize);
|
||||
mnodeGetDnodeEpsData(pEps, epsSize);
|
||||
|
||||
pMsg->rpcRsp.len = contLen;
|
||||
|
|
@ -711,7 +711,7 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessCreateDnodeMsg(SMnodeMsg *pMsg) {
|
||||
SCMCreateDnodeMsg *pCreate = pMsg->rpcMsg.pCont;
|
||||
SCreateDnodeMsg *pCreate = pMsg->rpcMsg.pCont;
|
||||
|
||||
if (strcmp(pMsg->pUser->user, TSDB_DEFAULT_USER) != 0) {
|
||||
return TSDB_CODE_MND_NO_RIGHTS;
|
||||
|
|
@ -721,7 +721,7 @@ static int32_t mnodeProcessCreateDnodeMsg(SMnodeMsg *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessDropDnodeMsg(SMnodeMsg *pMsg) {
|
||||
SCMDropDnodeMsg *pDrop = pMsg->rpcMsg.pCont;
|
||||
SDropDnodeMsg *pDrop = pMsg->rpcMsg.pCont;
|
||||
|
||||
if (strcmp(pMsg->pUser->user, TSDB_DEFAULT_USER) != 0) {
|
||||
return TSDB_CODE_MND_NO_RIGHTS;
|
||||
|
|
|
|||
|
|
@ -273,7 +273,7 @@ void mnodeGetMnodeInfos(void *mnodeInfos) {
|
|||
static int32_t mnodeSendCreateMnodeMsg(int32_t dnodeId, char *dnodeEp) {
|
||||
mDebug("dnode:%d, send create mnode msg to dnode %s", dnodeId, dnodeEp);
|
||||
|
||||
SMDCreateMnodeMsg *pCreate = rpcMallocCont(sizeof(SMDCreateMnodeMsg));
|
||||
SCreateMnodeMsg *pCreate = rpcMallocCont(sizeof(SCreateMnodeMsg));
|
||||
if (pCreate == NULL) {
|
||||
return TSDB_CODE_MND_OUT_OF_MEMORY;
|
||||
} else {
|
||||
|
|
@ -295,7 +295,7 @@ static int32_t mnodeSendCreateMnodeMsg(int32_t dnodeId, char *dnodeEp) {
|
|||
|
||||
SRpcMsg rpcMsg = {0};
|
||||
rpcMsg.pCont = pCreate;
|
||||
rpcMsg.contLen = sizeof(SMDCreateMnodeMsg);
|
||||
rpcMsg.contLen = sizeof(SCreateMnodeMsg);
|
||||
rpcMsg.msgType = TSDB_MSG_TYPE_MD_CREATE_MNODE;
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
|
|
|
|||
|
|
@ -280,7 +280,7 @@ static int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, voi
|
|||
}
|
||||
|
||||
// not thread safe, need optimized
|
||||
int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SCMHeartBeatMsg *pHBMsg) {
|
||||
int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pHBMsg) {
|
||||
pConn->numOfQueries = htonl(pHBMsg->numOfQueries);
|
||||
if (pConn->numOfQueries > 0) {
|
||||
if (pConn->pQueries == NULL) {
|
||||
|
|
@ -561,7 +561,7 @@ static int32_t mnodeProcessKillQueryMsg(SMnodeMsg *pMsg) {
|
|||
SUserObj *pUser = pMsg->pUser;
|
||||
if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
|
||||
|
||||
SCMKillQueryMsg *pKill = pMsg->rpcMsg.pCont;
|
||||
SKillQueryMsg *pKill = pMsg->rpcMsg.pCont;
|
||||
mInfo("kill query msg is received, queryId:%s", pKill->queryId);
|
||||
|
||||
const char delim = ':';
|
||||
|
|
@ -592,7 +592,7 @@ static int32_t mnodeProcessKillStreamMsg(SMnodeMsg *pMsg) {
|
|||
SUserObj *pUser = pMsg->pUser;
|
||||
if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
|
||||
|
||||
SCMKillQueryMsg *pKill = pMsg->rpcMsg.pCont;
|
||||
SKillQueryMsg *pKill = pMsg->rpcMsg.pCont;
|
||||
mInfo("kill stream msg is received, streamId:%s", pKill->queryId);
|
||||
|
||||
const char delim = ':';
|
||||
|
|
@ -623,7 +623,7 @@ static int32_t mnodeProcessKillConnectionMsg(SMnodeMsg *pMsg) {
|
|||
SUserObj *pUser = pMsg->pUser;
|
||||
if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
|
||||
|
||||
SCMKillConnMsg *pKill = pMsg->rpcMsg.pCont;
|
||||
SKillConnMsg *pKill = pMsg->rpcMsg.pCont;
|
||||
int32_t connId = atoi(pKill->queryId);
|
||||
SConnObj *pConn = taosCacheAcquireByKey(tsMnodeConnCache, &connId, sizeof(int32_t));
|
||||
if (pConn == NULL) {
|
||||
|
|
|
|||
|
|
@ -110,7 +110,7 @@ static char *mnodeGetShowType(int32_t showType) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessShowMsg(SMnodeMsg *pMsg) {
|
||||
SCMShowMsg *pShowMsg = pMsg->rpcMsg.pCont;
|
||||
SShowMsg *pShowMsg = pMsg->rpcMsg.pCont;
|
||||
if (pShowMsg->type >= TSDB_MGMT_TABLE_MAX) {
|
||||
return TSDB_CODE_MND_INVALID_MSG_TYPE;
|
||||
}
|
||||
|
|
@ -132,8 +132,8 @@ static int32_t mnodeProcessShowMsg(SMnodeMsg *pMsg) {
|
|||
return TSDB_CODE_MND_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
int32_t size = sizeof(SCMShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE;
|
||||
SCMShowRsp *pShowRsp = rpcMallocCont(size);
|
||||
int32_t size = sizeof(SShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE;
|
||||
SShowRsp *pShowRsp = rpcMallocCont(size);
|
||||
if (pShowRsp == NULL) {
|
||||
mnodeReleaseShowObj(pShow, true);
|
||||
return TSDB_CODE_MND_OUT_OF_MEMORY;
|
||||
|
|
@ -146,7 +146,7 @@ static int32_t mnodeProcessShowMsg(SMnodeMsg *pMsg) {
|
|||
|
||||
if (code == TSDB_CODE_SUCCESS) {
|
||||
pMsg->rpcRsp.rsp = pShowRsp;
|
||||
pMsg->rpcRsp.len = sizeof(SCMShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
|
||||
pMsg->rpcRsp.len = sizeof(SShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
|
||||
mnodeReleaseShowObj(pShow, false);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
} else {
|
||||
|
|
@ -232,12 +232,12 @@ static int32_t mnodeProcessRetrieveMsg(SMnodeMsg *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
|
||||
SCMHeartBeatRsp *pHBRsp = (SCMHeartBeatRsp *) rpcMallocCont(sizeof(SCMHeartBeatRsp));
|
||||
SHeartBeatRsp *pHBRsp = (SHeartBeatRsp *) rpcMallocCont(sizeof(SHeartBeatRsp));
|
||||
if (pHBRsp == NULL) {
|
||||
return TSDB_CODE_MND_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
SCMHeartBeatMsg *pHBMsg = pMsg->rpcMsg.pCont;
|
||||
SHeartBeatMsg *pHBMsg = pMsg->rpcMsg.pCont;
|
||||
SRpcConnInfo connInfo = {0};
|
||||
rpcGetConnInfo(pMsg->rpcMsg.handle, &connInfo);
|
||||
|
||||
|
|
@ -276,7 +276,7 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
|
|||
mnodeGetMnodeEpSetForShell(&pHBRsp->epSet);
|
||||
|
||||
pMsg->rpcRsp.rsp = pHBRsp;
|
||||
pMsg->rpcRsp.len = sizeof(SCMHeartBeatRsp);
|
||||
pMsg->rpcRsp.len = sizeof(SHeartBeatRsp);
|
||||
|
||||
mnodeReleaseConn(pConn);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
|
@ -356,7 +356,7 @@ connect_over:
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessUseMsg(SMnodeMsg *pMsg) {
|
||||
SCMUseDbMsg *pUseDbMsg = pMsg->rpcMsg.pCont;
|
||||
SUseDbMsg *pUseDbMsg = pMsg->rpcMsg.pCont;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pUseDbMsg->db);
|
||||
|
|
|
|||
|
|
@ -778,7 +778,7 @@ static int32_t mnodeProcessDropTableMsg(SMnodeMsg *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessTableMetaMsg(SMnodeMsg *pMsg) {
|
||||
SCMTableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
STableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
pInfo->createFlag = htons(pInfo->createFlag);
|
||||
mDebug("app:%p:%p, table:%s, table meta msg is received from thandle:%p, createFlag:%d", pMsg->rpcMsg.ahandle, pMsg,
|
||||
pInfo->tableId, pMsg->rpcMsg.handle, pInfo->createFlag);
|
||||
|
|
@ -915,8 +915,8 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
|
|||
SVgObj *pVgroup = mnodeGetVgroup(*pVgId);
|
||||
if (pVgroup == NULL) break;
|
||||
|
||||
SMDDropSTableMsg *pDrop = rpcMallocCont(sizeof(SMDDropSTableMsg));
|
||||
pDrop->contLen = htonl(sizeof(SMDDropSTableMsg));
|
||||
SDropSTableMsg *pDrop = rpcMallocCont(sizeof(SDropSTableMsg));
|
||||
pDrop->contLen = htonl(sizeof(SDropSTableMsg));
|
||||
pDrop->vgId = htonl(pVgroup->vgId);
|
||||
pDrop->uid = htobe64(pStable->uid);
|
||||
mnodeExtractTableName(pStable->info.tableId, pDrop->tableId);
|
||||
|
|
@ -924,7 +924,7 @@ static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg) {
|
|||
mInfo("app:%p:%p, stable:%s, send drop stable msg to vgId:%d", pMsg->rpcMsg.ahandle, pMsg, pStable->info.tableId,
|
||||
pVgroup->vgId);
|
||||
SRpcEpSet epSet = mnodeGetEpSetFromVgroup(pVgroup);
|
||||
SRpcMsg rpcMsg = {.pCont = pDrop, .contLen = sizeof(SMDDropSTableMsg), .msgType = TSDB_MSG_TYPE_MD_DROP_STABLE};
|
||||
SRpcMsg rpcMsg = {.pCont = pDrop, .contLen = sizeof(SDropSTableMsg), .msgType = TSDB_MSG_TYPE_MD_DROP_STABLE};
|
||||
dnodeSendMsgToDnode(&epSet, &rpcMsg);
|
||||
mnodeDecVgroupRef(pVgroup);
|
||||
}
|
||||
|
|
@ -1472,31 +1472,31 @@ static int32_t mnodeGetSuperTableMeta(SMnodeMsg *pMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
|
||||
SCMSTableVgroupMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
SSTableVgroupMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
int32_t numOfTable = htonl(pInfo->numOfTables);
|
||||
|
||||
// reserve space
|
||||
int32_t contLen = sizeof(SCMSTableVgroupRspMsg) + 32 * sizeof(SCMVgroupMsg) + sizeof(SVgroupsMsg);
|
||||
int32_t contLen = sizeof(SSTableVgroupRspMsg) + 32 * sizeof(SVgroupMsg) + sizeof(SVgroupsMsg);
|
||||
for (int32_t i = 0; i < numOfTable; ++i) {
|
||||
char *stableName = (char*)pInfo + sizeof(SCMSTableVgroupMsg) + (TSDB_TABLE_FNAME_LEN) * i;
|
||||
char *stableName = (char *)pInfo + sizeof(SSTableVgroupMsg) + (TSDB_TABLE_FNAME_LEN)*i;
|
||||
SSuperTableObj *pTable = mnodeGetSuperTable(stableName);
|
||||
if (pTable != NULL && pTable->vgHash != NULL) {
|
||||
contLen += (taosHashGetSize(pTable->vgHash) * sizeof(SCMVgroupMsg) + sizeof(SVgroupsMsg));
|
||||
contLen += (taosHashGetSize(pTable->vgHash) * sizeof(SVgroupMsg) + sizeof(SVgroupsMsg));
|
||||
}
|
||||
|
||||
mnodeDecTableRef(pTable);
|
||||
}
|
||||
|
||||
SCMSTableVgroupRspMsg *pRsp = rpcMallocCont(contLen);
|
||||
SSTableVgroupRspMsg *pRsp = rpcMallocCont(contLen);
|
||||
if (pRsp == NULL) {
|
||||
return TSDB_CODE_MND_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
pRsp->numOfTables = 0;
|
||||
char *msg = (char *)pRsp + sizeof(SCMSTableVgroupRspMsg);
|
||||
char *msg = (char *)pRsp + sizeof(SSTableVgroupRspMsg);
|
||||
|
||||
for (int32_t i = 0; i < numOfTable; ++i) {
|
||||
char * stableName = (char *)pInfo + sizeof(SCMSTableVgroupMsg) + (TSDB_TABLE_FNAME_LEN)*i;
|
||||
char * stableName = (char *)pInfo + sizeof(SSTableVgroupMsg) + (TSDB_TABLE_FNAME_LEN)*i;
|
||||
SSuperTableObj *pTable = mnodeGetSuperTable(stableName);
|
||||
if (pTable == NULL) {
|
||||
mError("app:%p:%p, stable:%s, not exist while get stable vgroup info", pMsg->rpcMsg.ahandle, pMsg, stableName);
|
||||
|
|
@ -1548,7 +1548,7 @@ static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *pMsg) {
|
|||
pVgroupMsg->numOfVgroups = htonl(vgSize);
|
||||
|
||||
// one table is done, try the next table
|
||||
msg += sizeof(SVgroupsMsg) + vgSize * sizeof(SCMVgroupMsg);
|
||||
msg += sizeof(SVgroupsMsg) + vgSize * sizeof(SVgroupMsg);
|
||||
pRsp->numOfTables++;
|
||||
}
|
||||
}
|
||||
|
|
@ -2146,7 +2146,7 @@ static int32_t mnodeDoGetChildTableMeta(SMnodeMsg *pMsg, STableMetaMsg *pMeta) {
|
|||
}
|
||||
|
||||
static int32_t mnodeAutoCreateChildTable(SMnodeMsg *pMsg) {
|
||||
SCMTableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
STableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
STagData *pTags = (STagData *)pInfo->tags;
|
||||
int32_t tagLen = htonl(pTags->dataLen);
|
||||
if (pTags->name[0] == 0) {
|
||||
|
|
@ -2307,7 +2307,7 @@ static SChildTableObj* mnodeGetTableByPos(int32_t vnode, int32_t tid) {
|
|||
static int32_t mnodeProcessTableCfgMsg(SMnodeMsg *pMsg) {
|
||||
return TSDB_CODE_COM_OPS_NOT_SUPPORT;
|
||||
#if 0
|
||||
SDMConfigTableMsg *pCfg = pMsg->rpcMsg.pCont;
|
||||
SConfigTableMsg *pCfg = pMsg->rpcMsg.pCont;
|
||||
pCfg->dnodeId = htonl(pCfg->dnodeId);
|
||||
pCfg->vgId = htonl(pCfg->vgId);
|
||||
pCfg->sid = htonl(pCfg->sid);
|
||||
|
|
@ -2469,7 +2469,7 @@ static void mnodeProcessAlterTableRsp(SRpcMsg *rpcMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *pMsg) {
|
||||
SCMMultiTableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
SMultiTableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
|
||||
pInfo->numOfTables = htonl(pInfo->numOfTables);
|
||||
|
||||
int32_t totalMallocLen = 4 * 1024 * 1024; // first malloc 4 MB, subsequent reallocation as twice
|
||||
|
|
@ -2705,7 +2705,7 @@ static int32_t mnodeRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessAlterTableMsg(SMnodeMsg *pMsg) {
|
||||
SCMAlterTableMsg *pAlter = pMsg->rpcMsg.pCont;
|
||||
SAlterTableMsg *pAlter = pMsg->rpcMsg.pCont;
|
||||
mDebug("app:%p:%p, table:%s, alter table msg is received from thandle:%p", pMsg->rpcMsg.ahandle, pMsg,
|
||||
pAlter->tableId, pMsg->rpcMsg.handle);
|
||||
|
||||
|
|
|
|||
|
|
@ -414,7 +414,7 @@ static int32_t mnodeProcessCreateUserMsg(SMnodeMsg *pMsg) {
|
|||
SUserObj *pOperUser = pMsg->pUser;
|
||||
|
||||
if (pOperUser->superAuth) {
|
||||
SCMCreateUserMsg *pCreate = pMsg->rpcMsg.pCont;
|
||||
SCreateUserMsg *pCreate = pMsg->rpcMsg.pCont;
|
||||
return mnodeCreateUser(pOperUser->pAcct, pCreate->user, pCreate->pass, pMsg);
|
||||
} else {
|
||||
mError("user:%s, no rights to create user", pOperUser->user);
|
||||
|
|
@ -426,7 +426,7 @@ static int32_t mnodeProcessAlterUserMsg(SMnodeMsg *pMsg) {
|
|||
int32_t code;
|
||||
SUserObj *pOperUser = pMsg->pUser;
|
||||
|
||||
SCMAlterUserMsg *pAlter = pMsg->rpcMsg.pCont;
|
||||
SAlterUserMsg *pAlter = pMsg->rpcMsg.pCont;
|
||||
SUserObj *pUser = mnodeGetUser(pAlter->user);
|
||||
if (pUser == NULL) {
|
||||
return TSDB_CODE_MND_INVALID_USER;
|
||||
|
|
@ -514,7 +514,7 @@ static int32_t mnodeProcessDropUserMsg(SMnodeMsg *pMsg) {
|
|||
int32_t code;
|
||||
SUserObj *pOperUser = pMsg->pUser;
|
||||
|
||||
SCMDropUserMsg *pDrop = pMsg->rpcMsg.pCont;
|
||||
SDropUserMsg *pDrop = pMsg->rpcMsg.pCont;
|
||||
SUserObj *pUser = mnodeGetUser(pDrop->user);
|
||||
if (pUser == NULL) {
|
||||
return TSDB_CODE_MND_INVALID_USER;
|
||||
|
|
@ -604,11 +604,11 @@ int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, cha
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessAuthMsg(SMnodeMsg *pMsg) {
|
||||
SDMAuthMsg *pAuthMsg = pMsg->rpcMsg.pCont;
|
||||
SDMAuthRsp *pAuthRsp = rpcMallocCont(sizeof(SDMAuthRsp));
|
||||
SAuthMsg *pAuthMsg = pMsg->rpcMsg.pCont;
|
||||
SAuthRsp *pAuthRsp = rpcMallocCont(sizeof(SAuthRsp));
|
||||
|
||||
pMsg->rpcRsp.rsp = pAuthRsp;
|
||||
pMsg->rpcRsp.len = sizeof(SDMAuthRsp);
|
||||
pMsg->rpcRsp.len = sizeof(SAuthRsp);
|
||||
|
||||
return mnodeRetriveAuth(pAuthMsg->user, &pAuthRsp->spi, &pAuthRsp->encrypt, pAuthRsp->secret, pAuthRsp->ckey);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -818,11 +818,11 @@ void mnodeRemoveTableFromVgroup(SVgObj *pVgroup, SChildTableObj *pTable) {
|
|||
}
|
||||
}
|
||||
|
||||
static SMDCreateVnodeMsg *mnodeBuildVnodeMsg(SVgObj *pVgroup) {
|
||||
static SCreateVnodeMsg *mnodeBuildVnodeMsg(SVgObj *pVgroup) {
|
||||
SDbObj *pDb = pVgroup->pDb;
|
||||
if (pDb == NULL) return NULL;
|
||||
|
||||
SMDCreateVnodeMsg *pVnode = rpcMallocCont(sizeof(SMDCreateVnodeMsg));
|
||||
SCreateVnodeMsg *pVnode = rpcMallocCont(sizeof(SCreateVnodeMsg));
|
||||
if (pVnode == NULL) return NULL;
|
||||
|
||||
strcpy(pVnode->db, pVgroup->dbName);
|
||||
|
|
@ -830,7 +830,7 @@ static SMDCreateVnodeMsg *mnodeBuildVnodeMsg(SVgObj *pVgroup) {
|
|||
//TODO: dynamic alloc tables in tsdb
|
||||
maxTables = MAX(10000, tsMaxTablePerVnode);
|
||||
|
||||
SMDVnodeCfg *pCfg = &pVnode->cfg;
|
||||
SVnodeCfg *pCfg = &pVnode->cfg;
|
||||
pCfg->vgId = htonl(pVgroup->vgId);
|
||||
pCfg->cfgVersion = htonl(pDb->cfgVersion);
|
||||
pCfg->cacheBlockSize = htonl(pDb->cfg.cacheBlockSize);
|
||||
|
|
@ -851,7 +851,7 @@ static SMDCreateVnodeMsg *mnodeBuildVnodeMsg(SVgObj *pVgroup) {
|
|||
pCfg->wals = 3;
|
||||
pCfg->quorum = pDb->cfg.quorum;
|
||||
|
||||
SMDVnodeDesc *pNodes = pVnode->nodes;
|
||||
SVnodeDesc *pNodes = pVnode->nodes;
|
||||
for (int32_t j = 0; j < pVgroup->numOfVnodes; ++j) {
|
||||
SDnodeObj *pDnode = pVgroup->vnodeGid[j].pDnode;
|
||||
if (pDnode != NULL) {
|
||||
|
|
@ -886,11 +886,11 @@ SRpcEpSet mnodeGetEpSetFromIp(char *ep) {
|
|||
}
|
||||
|
||||
static void mnodeSendAlterVnodeMsg(SVgObj *pVgroup, SRpcEpSet *epSet) {
|
||||
SMDAlterVnodeMsg *pAlter = mnodeBuildVnodeMsg(pVgroup);
|
||||
SAlterVnodeMsg *pAlter = mnodeBuildVnodeMsg(pVgroup);
|
||||
SRpcMsg rpcMsg = {
|
||||
.ahandle = NULL,
|
||||
.pCont = pAlter,
|
||||
.contLen = pAlter ? sizeof(SMDAlterVnodeMsg) : 0,
|
||||
.contLen = pAlter ? sizeof(SAlterVnodeMsg) : 0,
|
||||
.code = 0,
|
||||
.msgType = TSDB_MSG_TYPE_MD_ALTER_VNODE
|
||||
};
|
||||
|
|
@ -909,11 +909,11 @@ void mnodeSendAlterVgroupMsg(SVgObj *pVgroup) {
|
|||
}
|
||||
|
||||
static void mnodeSendCreateVnodeMsg(SVgObj *pVgroup, SRpcEpSet *epSet, void *ahandle) {
|
||||
SMDCreateVnodeMsg *pCreate = mnodeBuildVnodeMsg(pVgroup);
|
||||
SCreateVnodeMsg *pCreate = mnodeBuildVnodeMsg(pVgroup);
|
||||
SRpcMsg rpcMsg = {
|
||||
.ahandle = ahandle,
|
||||
.pCont = pCreate,
|
||||
.contLen = pCreate ? sizeof(SMDCreateVnodeMsg) : 0,
|
||||
.contLen = pCreate ? sizeof(SCreateVnodeMsg) : 0,
|
||||
.code = 0,
|
||||
.msgType = TSDB_MSG_TYPE_MD_CREATE_VNODE
|
||||
};
|
||||
|
|
@ -982,8 +982,8 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
|
|||
}
|
||||
}
|
||||
|
||||
static SMDDropVnodeMsg *mnodeBuildDropVnodeMsg(int32_t vgId) {
|
||||
SMDDropVnodeMsg *pDrop = rpcMallocCont(sizeof(SMDDropVnodeMsg));
|
||||
static SDropVnodeMsg *mnodeBuildDropVnodeMsg(int32_t vgId) {
|
||||
SDropVnodeMsg *pDrop = rpcMallocCont(sizeof(SDropVnodeMsg));
|
||||
if (pDrop == NULL) return NULL;
|
||||
|
||||
pDrop->vgId = htonl(vgId);
|
||||
|
|
@ -991,11 +991,11 @@ static SMDDropVnodeMsg *mnodeBuildDropVnodeMsg(int32_t vgId) {
|
|||
}
|
||||
|
||||
void mnodeSendDropVnodeMsg(int32_t vgId, SRpcEpSet *epSet, void *ahandle) {
|
||||
SMDDropVnodeMsg *pDrop = mnodeBuildDropVnodeMsg(vgId);
|
||||
SDropVnodeMsg *pDrop = mnodeBuildDropVnodeMsg(vgId);
|
||||
SRpcMsg rpcMsg = {
|
||||
.ahandle = ahandle,
|
||||
.pCont = pDrop,
|
||||
.contLen = pDrop ? sizeof(SMDDropVnodeMsg) : 0,
|
||||
.contLen = pDrop ? sizeof(SDropVnodeMsg) : 0,
|
||||
.code = 0,
|
||||
.msgType = TSDB_MSG_TYPE_MD_DROP_VNODE
|
||||
};
|
||||
|
|
@ -1044,7 +1044,7 @@ static void mnodeProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
|
|||
}
|
||||
|
||||
static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) {
|
||||
SDMConfigVnodeMsg *pCfg = pMsg->rpcMsg.pCont;
|
||||
SConfigVnodeMsg *pCfg = pMsg->rpcMsg.pCont;
|
||||
pCfg->dnodeId = htonl(pCfg->dnodeId);
|
||||
pCfg->vgId = htonl(pCfg->vgId);
|
||||
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ extern "C" {
|
|||
#endif
|
||||
|
||||
int32_t vnodeReadCfg(SVnodeObj *pVnode);
|
||||
int32_t vnodeWriteCfg(SMDCreateVnodeMsg *pVnodeCfg);
|
||||
int32_t vnodeWriteCfg(SCreateVnodeMsg *pVnodeCfg);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@
|
|||
#include "vnodeVersion.h"
|
||||
#include "vnodeCfg.h"
|
||||
|
||||
static void vnodeLoadCfg(SVnodeObj *pVnode, SMDCreateVnodeMsg* vnodeMsg) {
|
||||
static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) {
|
||||
strcpy(pVnode->db, vnodeMsg->db);
|
||||
pVnode->cfgVersion = vnodeMsg->cfg.cfgVersion;
|
||||
pVnode->tsdbCfg.cacheBlockSize = vnodeMsg->cfg.cacheBlockSize;
|
||||
|
|
@ -46,7 +46,7 @@ static void vnodeLoadCfg(SVnodeObj *pVnode, SMDCreateVnodeMsg* vnodeMsg) {
|
|||
pVnode->syncCfg.quorum = vnodeMsg->cfg.quorum;
|
||||
|
||||
for (int i = 0; i < pVnode->syncCfg.replica; ++i) {
|
||||
SMDVnodeDesc *node = &vnodeMsg->nodes[i];
|
||||
SVnodeDesc *node = &vnodeMsg->nodes[i];
|
||||
pVnode->syncCfg.nodeInfo[i].nodeId = node->nodeId;
|
||||
taosGetFqdnPortFromEp(node->nodeEp, pVnode->syncCfg.nodeInfo[i].nodeFqdn, &pVnode->syncCfg.nodeInfo[i].nodePort);
|
||||
pVnode->syncCfg.nodeInfo[i].nodePort += TSDB_PORT_SYNC;
|
||||
|
|
@ -67,7 +67,7 @@ int32_t vnodeReadCfg(SVnodeObj *pVnode) {
|
|||
cJSON * root = NULL;
|
||||
FILE * fp = NULL;
|
||||
bool nodeChanged = false;
|
||||
SMDCreateVnodeMsg vnodeMsg;
|
||||
SCreateVnodeMsg vnodeMsg;
|
||||
|
||||
char file[TSDB_FILENAME_LEN + 30] = {0};
|
||||
sprintf(file, "%s/vnode%d/config.json", tsVnodeDir, pVnode->vgId);
|
||||
|
|
@ -226,7 +226,7 @@ int32_t vnodeReadCfg(SVnodeObj *pVnode) {
|
|||
for (int i = 0; i < size; ++i) {
|
||||
cJSON *nodeInfo = cJSON_GetArrayItem(nodeInfos, i);
|
||||
if (nodeInfo == NULL) continue;
|
||||
SMDVnodeDesc *node = &vnodeMsg.nodes[i];
|
||||
SVnodeDesc *node = &vnodeMsg.nodes[i];
|
||||
|
||||
cJSON *nodeId = cJSON_GetObjectItem(nodeInfo, "nodeId");
|
||||
if (!nodeId || nodeId->type != cJSON_Number) {
|
||||
|
|
@ -266,7 +266,7 @@ PARSE_VCFG_ERROR:
|
|||
return ret;
|
||||
}
|
||||
|
||||
int32_t vnodeWriteCfg(SMDCreateVnodeMsg *pMsg) {
|
||||
int32_t vnodeWriteCfg(SCreateVnodeMsg *pMsg) {
|
||||
char file[TSDB_FILENAME_LEN + 30] = {0};
|
||||
sprintf(file, "%s/vnode%d/config.json", tsVnodeDir, pMsg->cfg.vgId);
|
||||
|
||||
|
|
@ -301,7 +301,7 @@ int32_t vnodeWriteCfg(SMDCreateVnodeMsg *pMsg) {
|
|||
len += snprintf(content + len, maxLen - len, " \"quorum\": %d,\n", pMsg->cfg.quorum);
|
||||
len += snprintf(content + len, maxLen - len, " \"nodeInfos\": [{\n");
|
||||
for (int32_t i = 0; i < pMsg->cfg.replications; i++) {
|
||||
SMDVnodeDesc *node = &pMsg->nodes[i];
|
||||
SVnodeDesc *node = &pMsg->nodes[i];
|
||||
dnodeUpdateEp(node->nodeId, node->nodeEp, NULL, NULL);
|
||||
len += snprintf(content + len, maxLen - len, " \"nodeId\": %d,\n", node->nodeId);
|
||||
len += snprintf(content + len, maxLen - len, " \"nodeEp\": \"%s\"\n", node->nodeEp);
|
||||
|
|
|
|||
|
|
@ -86,7 +86,7 @@ void vnodeCleanupResources() {
|
|||
syncCleanUp();
|
||||
}
|
||||
|
||||
int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
|
||||
int32_t vnodeCreate(SCreateVnodeMsg *pVnodeCfg) {
|
||||
int32_t code;
|
||||
|
||||
SVnodeObj *pVnode = vnodeAcquire(pVnodeCfg->cfg.vgId);
|
||||
|
|
@ -171,7 +171,7 @@ int32_t vnodeDrop(int32_t vgId) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t vnodeAlter(void *param, SMDCreateVnodeMsg *pVnodeCfg) {
|
||||
int32_t vnodeAlter(void *param, SCreateVnodeMsg *pVnodeCfg) {
|
||||
SVnodeObj *pVnode = param;
|
||||
|
||||
// vnode in non-ready state and still needs to return success instead of TSDB_CODE_VND_INVALID_STATUS
|
||||
|
|
@ -490,7 +490,7 @@ void *vnodeGetWal(void *pVnode) {
|
|||
return ((SVnodeObj *)pVnode)->wal;
|
||||
}
|
||||
|
||||
static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SDMStatusMsg *pStatus) {
|
||||
static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) {
|
||||
int64_t totalStorage = 0;
|
||||
int64_t compStorage = 0;
|
||||
int64_t pointsWritten = 0;
|
||||
|
|
@ -534,7 +534,7 @@ int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
|
|||
}
|
||||
|
||||
void vnodeBuildStatusMsg(void *param) {
|
||||
SDMStatusMsg *pStatus = param;
|
||||
SStatusMsg *pStatus = param;
|
||||
SHashMutableIterator *pIter = taosHashCreateIter(tsDnodeVnodesHash);
|
||||
|
||||
while (taosHashIterNext(pIter)) {
|
||||
|
|
@ -548,7 +548,7 @@ void vnodeBuildStatusMsg(void *param) {
|
|||
taosHashDestroyIter(pIter);
|
||||
}
|
||||
|
||||
void vnodeSetAccess(SDMVgroupAccess *pAccess, int32_t numOfVnodes) {
|
||||
void vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes) {
|
||||
for (int32_t i = 0; i < numOfVnodes; ++i) {
|
||||
pAccess[i].vgId = htonl(pAccess[i].vgId);
|
||||
SVnodeObj *pVnode = vnodeAcquire(pAccess[i].vgId);
|
||||
|
|
|
|||
|
|
@ -183,7 +183,7 @@ static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
|
|||
}
|
||||
|
||||
static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pRet) {
|
||||
SMDDropSTableMsg *pTable = pCont;
|
||||
SDropSTableMsg *pTable = pCont;
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
|
||||
vDebug("vgId:%d, stable:%s, start to drop", pVnode->vgId, pTable->tableId);
|
||||
|
|
|
|||
Loading…
Reference in New Issue