Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 55ec2aa0 编写于 作者: D dapan1121
浏览文件
操作

catalog async test

上级 a38fcf4e
隐藏空白更改
内联 并排
Showing with 227 addition and 136 deletion
include/libs/catalog/catalog.h
浏览文件 @ 55ec2aa0
...@@ -64,7 +64,7 @@ typedef struct SCatalogReq { ...@@ -64,7 +64,7 @@ typedef struct SCatalogReq {
} SCatalogReq; } SCatalogReq;
typedef struct SMetaData { typedef struct SMetaData {
SArray *pTableMeta; // SArray<STableMeta> SArray *pTableMeta; // SArray<STableMeta*>
SArray *pDbVgroup; // SArray<SArray<SVgroupInfo>*> SArray *pDbVgroup; // SArray<SArray<SVgroupInfo>*>
SArray *pTableHash; // SArray<SVgroupInfo> SArray *pTableHash; // SArray<SVgroupInfo>
SArray *pUdfList; // SArray<SFuncInfo> SArray *pUdfList; // SArray<SFuncInfo>
...@@ -248,6 +248,8 @@ int32_t catalogGetTableHashVgroup(SCatalog* pCatalog, void * pTransporter, const ...@@ -248,6 +248,8 @@ int32_t catalogGetTableHashVgroup(SCatalog* pCatalog, void * pTransporter, const
*/ */
int32_t catalogGetAllMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp); int32_t catalogGetAllMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param, int64_t* jobId);
int32_t catalogGetQnodeList(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList); int32_t catalogGetQnodeList(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList);
int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num); int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
...@@ -267,6 +269,9 @@ int32_t catalogChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const ...@@ -267,6 +269,9 @@ int32_t catalogChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const
int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth); int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth);
int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId);
/** /**
* Destroy catalog and relase all resources * Destroy catalog and relase all resources
*/ */
......
include/libs/qcom/query.h
浏览文件 @ 55ec2aa0
...@@ -180,7 +180,7 @@ char* jobTaskStatusStr(int32_t status); ...@@ -180,7 +180,7 @@ char* jobTaskStatusStr(int32_t status);
SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name); SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name);
extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char** msg, int32_t msgSize, int32_t* msgLen); extern int32_t (*queryBuildMsg[TDMT_MAX])(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallocFp)(int32_t));
extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t msgSize); extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t msgSize);
#define SET_META_TYPE_NULL(t) (t) = META_TYPE_NULL_TABLE #define SET_META_TYPE_NULL(t) (t) = META_TYPE_NULL_TABLE
......
source/libs/catalog/inc/catalogInt.h
浏览文件 @ 55ec2aa0
...@@ -171,7 +171,7 @@ typedef struct SCtgJob { ...@@ -171,7 +171,7 @@ typedef struct SCtgJob {
uint64_t queryId; uint64_t queryId;
SCatalog* pCtg; SCatalog* pCtg;
void* pTrans; void* pTrans;
const SEpSet* pMgmtEps; SEpSet pMgmtEps;
void* userParam; void* userParam;
catalogCallback userFp; catalogCallback userFp;
int32_t tbMetaNum; int32_t tbMetaNum;
......
source/libs/catalog/src/catalog.c
浏览文件 @ 55ec2aa0
...@@ -327,28 +327,31 @@ int32_t ctgChkAuth(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const c ...@@ -327,28 +327,31 @@ int32_t ctgChkAuth(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const c
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SGetUserAuthRsp authRsp = {0}; SGetUserAuthRsp* authRsp = taosMemoryCalloc(1, sizeof(SGetUserAuthRsp));
CTG_ERR_RET(ctgGetUserDbAuthFromMnode(CTG_PARAMS_LIST(), user, &authRsp, NULL)); if (NULL == authRsp) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
CTG_ERR_RET(ctgGetUserDbAuthFromMnode(CTG_PARAMS_LIST(), user, authRsp, NULL));
if (authRsp.superAuth) { if (authRsp->superAuth) {
*pass = true; *pass = true;
goto _return; goto _return;
} }
if (authRsp.createdDbs && taosHashGet(authRsp.createdDbs, dbFName, strlen(dbFName))) { if (authRsp->createdDbs && taosHashGet(authRsp->createdDbs, dbFName, strlen(dbFName))) {
*pass = true; *pass = true;
goto _return; goto _return;
} }
if (type == AUTH_TYPE_READ && authRsp.readDbs && taosHashGet(authRsp.readDbs, dbFName, strlen(dbFName))) { if (type == AUTH_TYPE_READ && authRsp->readDbs && taosHashGet(authRsp->readDbs, dbFName, strlen(dbFName))) {
*pass = true; *pass = true;
} else if (type == AUTH_TYPE_WRITE && authRsp.writeDbs && taosHashGet(authRsp.writeDbs, dbFName, strlen(dbFName))) { } else if (type == AUTH_TYPE_WRITE && authRsp->writeDbs && taosHashGet(authRsp->writeDbs, dbFName, strlen(dbFName))) {
*pass = true; *pass = true;
} }
_return: _return:
ctgPutUpdateUserToQueue(pCtg, &authRsp, false); ctgPutUpdateUserToQueue(pCtg, authRsp, false);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
......
source/libs/catalog/src/ctgAsync.c
浏览文件 @ 55ec2aa0
...@@ -21,173 +21,189 @@ ...@@ -21,173 +21,189 @@
#include "tref.h" #include "tref.h"
int32_t ctgInitGetTbMetaTask(SCtgJob *pJob, int32_t taskIdx, SName *name) { int32_t ctgInitGetTbMetaTask(SCtgJob *pJob, int32_t taskIdx, SName *name) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
pTask->type = CTG_TASK_GET_TB_META; task.type = CTG_TASK_GET_TB_META;
pTask->taskId = taskIdx; task.taskId = taskIdx;
pTask->pJob = pJob; task.pJob = pJob;
pTask->taskCtx = taosMemoryCalloc(1, sizeof(SCtgTbMetaCtx)); task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgTbMetaCtx));
if (NULL == pTask->taskCtx) { if (NULL == task.taskCtx) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
SCtgTbMetaCtx* ctx = pTask->taskCtx; SCtgTbMetaCtx* ctx = task.taskCtx;
ctx->pName = taosMemoryMalloc(sizeof(*name)); ctx->pName = taosMemoryMalloc(sizeof(*name));
if (NULL == ctx->pName) { if (NULL == ctx->pName) {
taosMemoryFree(pTask->taskCtx); taosMemoryFree(task.taskCtx);
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
memcpy(ctx->pName, name, sizeof(*name)); memcpy(ctx->pName, name, sizeof(*name));
ctx->flag = CTG_FLAG_UNKNOWN_STB; ctx->flag = CTG_FLAG_UNKNOWN_STB;
qDebug("QID:%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, pTask->type, name->tname); taosArrayPush(pJob->pTasks, &task);
qDebug("QID:%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, task.type, name->tname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgInitGetDbVgTask(SCtgJob *pJob, int32_t taskIdx, char *dbFName) { int32_t ctgInitGetDbVgTask(SCtgJob *pJob, int32_t taskIdx, char *dbFName) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
pTask->type = CTG_TASK_GET_DB_VGROUP; task.type = CTG_TASK_GET_DB_VGROUP;
pTask->taskId = taskIdx; task.taskId = taskIdx;
pTask->pJob = pJob; task.pJob = pJob;
pTask->taskCtx = taosMemoryCalloc(1, sizeof(SCtgDbVgCtx)); task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgDbVgCtx));
if (NULL == pTask->taskCtx) { if (NULL == task.taskCtx) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
SCtgDbVgCtx* ctx = pTask->taskCtx; SCtgDbVgCtx* ctx = task.taskCtx;
memcpy(ctx->dbFName, dbFName, sizeof(ctx->dbFName)); memcpy(ctx->dbFName, dbFName, sizeof(ctx->dbFName));
qDebug("QID:%" PRIx64 " task %d type %d initialized, dbFName:%s", pJob->queryId, taskIdx, pTask->type, dbFName); taosArrayPush(pJob->pTasks, &task);
qDebug("QID:%" PRIx64 " task %d type %d initialized, dbFName:%s", pJob->queryId, taskIdx, task.type, dbFName);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgInitGetDbCfgTask(SCtgJob *pJob, int32_t taskIdx, char *dbFName) { int32_t ctgInitGetDbCfgTask(SCtgJob *pJob, int32_t taskIdx, char *dbFName) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
pTask->type = CTG_TASK_GET_DB_CFG; task.type = CTG_TASK_GET_DB_CFG;
pTask->taskId = taskIdx; task.taskId = taskIdx;
pTask->pJob = pJob; task.pJob = pJob;
pTask->taskCtx = taosMemoryCalloc(1, sizeof(SCtgDbCfgCtx)); task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgDbCfgCtx));
if (NULL == pTask->taskCtx) { if (NULL == task.taskCtx) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
SCtgDbCfgCtx* ctx = pTask->taskCtx; SCtgDbCfgCtx* ctx = task.taskCtx;
memcpy(ctx->dbFName, dbFName, sizeof(ctx->dbFName)); memcpy(ctx->dbFName, dbFName, sizeof(ctx->dbFName));
qDebug("QID:%" PRIx64 " task %d type %d initialized, dbFName:%s", pJob->queryId, taskIdx, pTask->type, dbFName); taosArrayPush(pJob->pTasks, &task);
qDebug("QID:%" PRIx64 " task %d type %d initialized, dbFName:%s", pJob->queryId, taskIdx, task.type, dbFName);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgInitGetTbHashTask(SCtgJob *pJob, int32_t taskIdx, SName *name) { int32_t ctgInitGetTbHashTask(SCtgJob *pJob, int32_t taskIdx, SName *name) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
pTask->type = CTG_TASK_GET_TB_HASH; task.type = CTG_TASK_GET_TB_HASH;
pTask->taskId = taskIdx; task.taskId = taskIdx;
pTask->pJob = pJob; task.pJob = pJob;
pTask->taskCtx = taosMemoryCalloc(1, sizeof(SCtgTbHashCtx)); task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgTbHashCtx));
if (NULL == pTask->taskCtx) { if (NULL == task.taskCtx) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
SCtgTbHashCtx* ctx = pTask->taskCtx; SCtgTbHashCtx* ctx = task.taskCtx;
ctx->pName = taosMemoryMalloc(sizeof(*name)); ctx->pName = taosMemoryMalloc(sizeof(*name));
if (NULL == ctx->pName) { if (NULL == ctx->pName) {
taosMemoryFree(pTask->taskCtx); taosMemoryFree(task.taskCtx);
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
memcpy(ctx->pName, name, sizeof(*name)); memcpy(ctx->pName, name, sizeof(*name));
tNameGetFullDbName(ctx->pName, ctx->dbFName); tNameGetFullDbName(ctx->pName, ctx->dbFName);
qDebug("QID:%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, pTask->type, name->tname); taosArrayPush(pJob->pTasks, &task);
qDebug("QID:%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, task.type, name->tname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgInitGetQnodeTask(SCtgJob *pJob, int32_t taskIdx) { int32_t ctgInitGetQnodeTask(SCtgJob *pJob, int32_t taskIdx) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
task.type = CTG_TASK_GET_QNODE;
task.taskId = taskIdx;
task.pJob = pJob;
task.taskCtx = NULL;
pTask->type = CTG_TASK_GET_QNODE; taosArrayPush(pJob->pTasks, &task);
pTask->taskId = taskIdx;
pTask->pJob = pJob;
pTask->taskCtx = NULL;
qDebug("QID:%" PRIx64 " task %d type %d initialized", pJob->queryId, taskIdx, pTask->type); qDebug("QID:%" PRIx64 " task %d type %d initialized", pJob->queryId, taskIdx, task.type);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgInitGetIndexTask(SCtgJob *pJob, int32_t taskIdx, char *name) { int32_t ctgInitGetIndexTask(SCtgJob *pJob, int32_t taskIdx, char *name) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
pTask->type = CTG_TASK_GET_INDEX; task.type = CTG_TASK_GET_INDEX;
pTask->taskId = taskIdx; task.taskId = taskIdx;
pTask->pJob = pJob; task.pJob = pJob;
pTask->taskCtx = taosMemoryCalloc(1, sizeof(SCtgIndexCtx)); task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgIndexCtx));
if (NULL == pTask->taskCtx) { if (NULL == task.taskCtx) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
SCtgIndexCtx* ctx = pTask->taskCtx; SCtgIndexCtx* ctx = task.taskCtx;
strcpy(ctx->indexFName, name); strcpy(ctx->indexFName, name);
qDebug("QID:%" PRIx64 " task %d type %d initialized, indexFName:%s", pJob->queryId, taskIdx, pTask->type, name); taosArrayPush(pJob->pTasks, &task);
qDebug("QID:%" PRIx64 " task %d type %d initialized, indexFName:%s", pJob->queryId, taskIdx, task.type, name);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgInitGetUdfTask(SCtgJob *pJob, int32_t taskIdx, char *name) { int32_t ctgInitGetUdfTask(SCtgJob *pJob, int32_t taskIdx, char *name) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
pTask->type = CTG_TASK_GET_UDF; task.type = CTG_TASK_GET_UDF;
pTask->taskId = taskIdx; task.taskId = taskIdx;
pTask->pJob = pJob; task.pJob = pJob;
pTask->taskCtx = taosMemoryCalloc(1, sizeof(SCtgUdfCtx)); task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgUdfCtx));
if (NULL == pTask->taskCtx) { if (NULL == task.taskCtx) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
SCtgUdfCtx* ctx = pTask->taskCtx; SCtgUdfCtx* ctx = task.taskCtx;
strcpy(ctx->udfName, name); strcpy(ctx->udfName, name);
qDebug("QID:%" PRIx64 " task %d type %d initialized, udfName:%s", pJob->queryId, taskIdx, pTask->type, name); taosArrayPush(pJob->pTasks, &task);
qDebug("QID:%" PRIx64 " task %d type %d initialized, udfName:%s", pJob->queryId, taskIdx, task.type, name);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgInitGetUserTask(SCtgJob *pJob, int32_t taskIdx, SUserAuthInfo *user) { int32_t ctgInitGetUserTask(SCtgJob *pJob, int32_t taskIdx, SUserAuthInfo *user) {
SCtgTask *pTask = taosArrayGet(pJob->pTasks, taskIdx); SCtgTask task = {0};
pTask->type = CTG_TASK_GET_USER; task.type = CTG_TASK_GET_USER;
pTask->taskId = taskIdx; task.taskId = taskIdx;
pTask->pJob = pJob; task.pJob = pJob;
pTask->taskCtx = taosMemoryCalloc(1, sizeof(SCtgUserCtx)); task.taskCtx = taosMemoryCalloc(1, sizeof(SCtgUserCtx));
if (NULL == pTask->taskCtx) { if (NULL == task.taskCtx) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
SCtgUserCtx* ctx = pTask->taskCtx; SCtgUserCtx* ctx = task.taskCtx;
memcpy(&ctx->user, user, sizeof(*user)); memcpy(&ctx->user, user, sizeof(*user));
qDebug("QID:%" PRIx64 " task %d type %d initialized, user:%s", pJob->queryId, taskIdx, pTask->type, user->user); taosArrayPush(pJob->pTasks, &task);
qDebug("QID:%" PRIx64 " task %d type %d initialized, user:%s", pJob->queryId, taskIdx, task.type, user->user);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -222,7 +238,7 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq* ...@@ -222,7 +238,7 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
pJob->userFp = fp; pJob->userFp = fp;
pJob->pCtg = pCtg; pJob->pCtg = pCtg;
pJob->pTrans = pTrans; pJob->pTrans = pTrans;
pJob->pMgmtEps = pMgmtEps; pJob->pMgmtEps = *pMgmtEps;
pJob->userParam = param; pJob->userParam = param;
pJob->tbMetaNum = tbMetaNum; pJob->tbMetaNum = tbMetaNum;
...@@ -303,15 +319,13 @@ _return: ...@@ -303,15 +319,13 @@ _return:
int32_t ctgDumpTbMetaRes(SCtgTask* pTask) { int32_t ctgDumpTbMetaRes(SCtgTask* pTask) {
SCtgJob* pJob = pTask->pJob; SCtgJob* pJob = pTask->pJob;
if (NULL == pJob->jobRes.pTableMeta) { if (NULL == pJob->jobRes.pTableMeta) {
pJob->jobRes.pTableMeta = taosArrayInit(pJob->tbMetaNum, sizeof(STableMeta)); pJob->jobRes.pTableMeta = taosArrayInit(pJob->tbMetaNum, POINTER_BYTES);
if (NULL == pJob->jobRes.pTableMeta) { if (NULL == pJob->jobRes.pTableMeta) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
} }
taosArrayPush(pJob->jobRes.pTableMeta, pTask->res); taosArrayPush(pJob->jobRes.pTableMeta, &pTask->res);
taosMemoryFreeClear(pTask->res);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -340,7 +354,7 @@ int32_t ctgDumpTbHashRes(SCtgTask* pTask) { ...@@ -340,7 +354,7 @@ int32_t ctgDumpTbHashRes(SCtgTask* pTask) {
} }
} }
taosArrayPush(pJob->jobRes.pTableHash, &pTask->res); taosArrayPush(pJob->jobRes.pTableHash, pTask->res);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -376,7 +390,7 @@ int32_t ctgDumpDbCfgRes(SCtgTask* pTask) { ...@@ -376,7 +390,7 @@ int32_t ctgDumpDbCfgRes(SCtgTask* pTask) {
} }
} }
taosArrayPush(pJob->jobRes.pDbCfg, &pTask->res); taosArrayPush(pJob->jobRes.pDbCfg, pTask->res);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -451,7 +465,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf * ...@@ -451,7 +465,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx; SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
switch (reqType) { switch (reqType) {
case TDMT_MND_USE_DB: { case TDMT_MND_USE_DB: {
...@@ -529,7 +543,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf * ...@@ -529,7 +543,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
taosMemoryFreeClear(pOut->tbMeta); taosMemoryFreeClear(pOut->tbMeta);
CTG_ERR_JRET(ctgGetTbMetaFromMnode(CTG_PARAMS_LIST(), ctx->pName, NULL, pTask)); CTG_RET(ctgGetTbMetaFromMnode(CTG_PARAMS_LIST(), ctx->pName, NULL, pTask));
} else if (CTG_IS_META_BOTH(pOut->metaType)) { } else if (CTG_IS_META_BOTH(pOut->metaType)) {
int32_t exist = 0; int32_t exist = 0;
if (!CTG_FLAG_IS_FORCE_UPDATE(ctx->flag)) { if (!CTG_FLAG_IS_FORCE_UPDATE(ctx->flag)) {
...@@ -538,7 +552,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf * ...@@ -538,7 +552,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
if (0 == exist) { if (0 == exist) {
TSWAP(pTask->msgCtx.lastOut, pTask->msgCtx.out); TSWAP(pTask->msgCtx.lastOut, pTask->msgCtx.out);
CTG_ERR_JRET(ctgGetTbMetaFromMnodeImpl(CTG_PARAMS_LIST(), pOut->dbFName, pOut->tbName, NULL, pTask)); CTG_RET(ctgGetTbMetaFromMnodeImpl(CTG_PARAMS_LIST(), pOut->dbFName, pOut->tbName, NULL, pTask));
} else { } else {
taosMemoryFreeClear(pOut->tbMeta); taosMemoryFreeClear(pOut->tbMeta);
...@@ -598,7 +612,7 @@ int32_t ctgHandleGetDbVgRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pM ...@@ -598,7 +612,7 @@ int32_t ctgHandleGetDbVgRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pM
SCtgDbVgCtx* ctx = (SCtgDbVgCtx*)pTask->taskCtx; SCtgDbVgCtx* ctx = (SCtgDbVgCtx*)pTask->taskCtx;
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
switch (reqType) { switch (reqType) {
case TDMT_MND_USE_DB: { case TDMT_MND_USE_DB: {
...@@ -632,7 +646,7 @@ int32_t ctgHandleGetTbHashRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf * ...@@ -632,7 +646,7 @@ int32_t ctgHandleGetTbHashRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
SCtgTbHashCtx* ctx = (SCtgTbHashCtx*)pTask->taskCtx; SCtgTbHashCtx* ctx = (SCtgTbHashCtx*)pTask->taskCtx;
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
switch (reqType) { switch (reqType) {
case TDMT_MND_USE_DB: { case TDMT_MND_USE_DB: {
...@@ -724,7 +738,7 @@ int32_t ctgHandleGetUserRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pM ...@@ -724,7 +738,7 @@ int32_t ctgHandleGetUserRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pM
SCtgUserCtx* ctx = (SCtgUserCtx*)pTask->taskCtx; SCtgUserCtx* ctx = (SCtgUserCtx*)pTask->taskCtx;
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
bool pass = false; bool pass = false;
SGetUserAuthRsp* pOut = (SGetUserAuthRsp*)pTask->msgCtx.out; SGetUserAuthRsp* pOut = (SGetUserAuthRsp*)pTask->msgCtx.out;
...@@ -766,7 +780,7 @@ _return: ...@@ -766,7 +780,7 @@ _return:
int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) { int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
int32_t code = 0; int32_t code = 0;
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx; SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
...@@ -788,7 +802,7 @@ int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) { ...@@ -788,7 +802,7 @@ int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) {
tNameGetFullDbName(ctx->pName, dbFName); tNameGetFullDbName(ctx->pName, dbFName);
CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache)); CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));
if (NULL == dbCache) { if (dbCache) {
SVgroupInfo vgInfo = {0}; SVgroupInfo vgInfo = {0};
CTG_ERR_RET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, ctx->pName, &vgInfo)); CTG_ERR_RET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, ctx->pName, &vgInfo));
...@@ -817,7 +831,7 @@ _return: ...@@ -817,7 +831,7 @@ _return:
int32_t ctgLaunchGetTbMetaTask(SCtgTask *pTask) { int32_t ctgLaunchGetTbMetaTask(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
CTG_ERR_RET(ctgGetTbMetaFromCache(CTG_PARAMS_LIST(), (SCtgTbMetaCtx*)pTask->taskCtx, (STableMeta**)&pTask->res)); CTG_ERR_RET(ctgGetTbMetaFromCache(CTG_PARAMS_LIST(), (SCtgTbMetaCtx*)pTask->taskCtx, (STableMeta**)&pTask->res));
if (pTask->res) { if (pTask->res) {
...@@ -834,7 +848,7 @@ int32_t ctgLaunchGetDbVgTask(SCtgTask *pTask) { ...@@ -834,7 +848,7 @@ int32_t ctgLaunchGetDbVgTask(SCtgTask *pTask) {
int32_t code = 0; int32_t code = 0;
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
SCtgDbVgCtx* pCtx = (SCtgDbVgCtx*)pTask->taskCtx; SCtgDbVgCtx* pCtx = (SCtgDbVgCtx*)pTask->taskCtx;
...@@ -866,7 +880,7 @@ int32_t ctgLaunchGetTbHashTask(SCtgTask *pTask) { ...@@ -866,7 +880,7 @@ int32_t ctgLaunchGetTbHashTask(SCtgTask *pTask) {
int32_t code = 0; int32_t code = 0;
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
SCtgTbHashCtx* pCtx = (SCtgTbHashCtx*)pTask->taskCtx; SCtgTbHashCtx* pCtx = (SCtgTbHashCtx*)pTask->taskCtx;
...@@ -901,7 +915,7 @@ _return: ...@@ -901,7 +915,7 @@ _return:
int32_t ctgLaunchGetQnodeTask(SCtgTask *pTask) { int32_t ctgLaunchGetQnodeTask(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
CTG_ERR_RET(ctgGetQnodeListFromMnode(CTG_PARAMS_LIST(), NULL, pTask)); CTG_ERR_RET(ctgGetQnodeListFromMnode(CTG_PARAMS_LIST(), NULL, pTask));
...@@ -911,7 +925,7 @@ int32_t ctgLaunchGetQnodeTask(SCtgTask *pTask) { ...@@ -911,7 +925,7 @@ int32_t ctgLaunchGetQnodeTask(SCtgTask *pTask) {
int32_t ctgLaunchGetDbCfgTask(SCtgTask *pTask) { int32_t ctgLaunchGetDbCfgTask(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
SCtgDbCfgCtx* pCtx = (SCtgDbCfgCtx*)pTask->taskCtx; SCtgDbCfgCtx* pCtx = (SCtgDbCfgCtx*)pTask->taskCtx;
CTG_ERR_RET(ctgGetDBCfgFromMnode(CTG_PARAMS_LIST(), pCtx->dbFName, NULL, pTask)); CTG_ERR_RET(ctgGetDBCfgFromMnode(CTG_PARAMS_LIST(), pCtx->dbFName, NULL, pTask));
...@@ -922,7 +936,7 @@ int32_t ctgLaunchGetDbCfgTask(SCtgTask *pTask) { ...@@ -922,7 +936,7 @@ int32_t ctgLaunchGetDbCfgTask(SCtgTask *pTask) {
int32_t ctgLaunchGetIndexTask(SCtgTask *pTask) { int32_t ctgLaunchGetIndexTask(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
SCtgIndexCtx* pCtx = (SCtgIndexCtx*)pTask->taskCtx; SCtgIndexCtx* pCtx = (SCtgIndexCtx*)pTask->taskCtx;
CTG_ERR_RET(ctgGetIndexInfoFromMnode(CTG_PARAMS_LIST(), pCtx->indexFName, NULL, pTask)); CTG_ERR_RET(ctgGetIndexInfoFromMnode(CTG_PARAMS_LIST(), pCtx->indexFName, NULL, pTask));
...@@ -933,7 +947,7 @@ int32_t ctgLaunchGetIndexTask(SCtgTask *pTask) { ...@@ -933,7 +947,7 @@ int32_t ctgLaunchGetIndexTask(SCtgTask *pTask) {
int32_t ctgLaunchGetUdfTask(SCtgTask *pTask) { int32_t ctgLaunchGetUdfTask(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
SCtgUdfCtx* pCtx = (SCtgUdfCtx*)pTask->taskCtx; SCtgUdfCtx* pCtx = (SCtgUdfCtx*)pTask->taskCtx;
CTG_ERR_RET(ctgGetUdfInfoFromMnode(CTG_PARAMS_LIST(), pCtx->udfName, NULL, pTask)); CTG_ERR_RET(ctgGetUdfInfoFromMnode(CTG_PARAMS_LIST(), pCtx->udfName, NULL, pTask));
...@@ -944,7 +958,7 @@ int32_t ctgLaunchGetUdfTask(SCtgTask *pTask) { ...@@ -944,7 +958,7 @@ int32_t ctgLaunchGetUdfTask(SCtgTask *pTask) {
int32_t ctgLaunchGetUserTask(SCtgTask *pTask) { int32_t ctgLaunchGetUserTask(SCtgTask *pTask) {
SCatalog* pCtg = pTask->pJob->pCtg; SCatalog* pCtg = pTask->pJob->pCtg;
void *pTrans = pTask->pJob->pTrans; void *pTrans = pTask->pJob->pTrans;
const SEpSet* pMgmtEps = pTask->pJob->pMgmtEps; const SEpSet* pMgmtEps = &pTask->pJob->pMgmtEps;
SCtgUserCtx* pCtx = (SCtgUserCtx*)pTask->taskCtx; SCtgUserCtx* pCtx = (SCtgUserCtx*)pTask->taskCtx;
bool inCache = false; bool inCache = false;
bool pass = false; bool pass = false;
......
source/libs/catalog/src/ctgCache.c
浏览文件 @ 55ec2aa0
...@@ -248,7 +248,7 @@ int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** ...@@ -248,7 +248,7 @@ int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta**
ctgAcquireDBCache(pCtg, dbFName, &dbCache); ctgAcquireDBCache(pCtg, dbFName, &dbCache);
if (NULL == dbCache) { if (NULL == dbCache) {
ctgDebug("db %s not in cache", ctx->pName->tname); ctgDebug("db %d.%s not in cache", ctx->pName->acctId, ctx->pName->dbname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -716,6 +716,8 @@ int32_t ctgPutUpdateUserToQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syn ...@@ -716,6 +716,8 @@ int32_t ctgPutUpdateUserToQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syn
CTG_ERR_JRET(ctgPushAction(pCtg, &action)); CTG_ERR_JRET(ctgPushAction(pCtg, &action));
taosMemoryFree(pAuth);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
_return: _return:
......
source/libs/catalog/src/ctgDbg.c
浏览文件 @ 55ec2aa0
...@@ -25,9 +25,9 @@ void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) { ...@@ -25,9 +25,9 @@ void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) {
ASSERT(*(int32_t*)param == 1); ASSERT(*(int32_t*)param == 1);
taosMemoryFree(param); taosMemoryFree(param);
ctgDebug("async call result: %s", tstrerror(code)); qDebug("async call result: %s", tstrerror(code));
if (NULL == pResult) { if (NULL == pResult) {
ctgDebug("empty meta result"); qDebug("empty meta result");
return; return;
} }
...@@ -36,25 +36,24 @@ void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) { ...@@ -36,25 +36,24 @@ void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) {
if (pResult->pTableMeta && taosArrayGetSize(pResult->pTableMeta) > 0) { if (pResult->pTableMeta && taosArrayGetSize(pResult->pTableMeta) > 0) {
num = taosArrayGetSize(pResult->pTableMeta); num = taosArrayGetSize(pResult->pTableMeta);
for (int32_t i = 0; i < num; ++i) { for (int32_t i = 0; i < num; ++i) {
STableMeta *p = taosArrayGet(pResult->pTableMeta, i); STableMeta *p = *(STableMeta **)taosArrayGet(pResult->pTableMeta, i);
STableComInfo *c = &p->tableInfo; STableComInfo *c = &p->tableInfo;
if (TSDB_CHILD_TABLE == p->tableType) { if (TSDB_CHILD_TABLE == p->tableType) {
ctgDebug("table meta: type:%d, vgId:%d, uid:%" PRIx64 ",suid:%" PRIx64, p->tableType, p->vgId, p->uid, p->suid); qDebug("table meta: type:%d, vgId:%d, uid:%" PRIx64 ",suid:%" PRIx64, p->tableType, p->vgId, p->uid, p->suid);
return;
} else { } else {
ctgDebug("table meta: type:%d, vgId:%d, uid:%" PRIx64 ",suid:%" PRIx64 ",sv:%d, tv:%d, tagNum:%d, precision:%d, colNum:%d, rowSize:%d", qDebug("table meta: type:%d, vgId:%d, uid:%" PRIx64 ",suid:%" PRIx64 ",sv:%d, tv:%d, tagNum:%d, precision:%d, colNum:%d, rowSize:%d",
p->tableType, p->vgId, p->uid, p->suid, p->sversion, p->tversion, c->numOfTags, c->precision, c->numOfColumns, c->rowSize); p->tableType, p->vgId, p->uid, p->suid, p->sversion, p->tversion, c->numOfTags, c->precision, c->numOfColumns, c->rowSize);
} }
int32_t colNum = c->numOfColumns + c->numOfTags; int32_t colNum = c->numOfColumns + c->numOfTags;
for (int32_t j = 0; j < colNum; ++j) { for (int32_t j = 0; j < colNum; ++j) {
SSchema *s = &p->schema[j]; SSchema *s = &p->schema[j];
ctgDebug("[%d] name:%s, type:%d, colId:%d, bytes:%d", j, s->name, s->type, s->colId, s->bytes); qDebug("[%d] name:%s, type:%d, colId:%d, bytes:%d", j, s->name, s->type, s->colId, s->bytes);
} }
} }
} else { } else {
ctgDebug("empty table meta"); qDebug("empty table meta");
} }
if (pResult->pDbVgroup && taosArrayGetSize(pResult->pDbVgroup) > 0) { if (pResult->pDbVgroup && taosArrayGetSize(pResult->pDbVgroup) > 0) {
...@@ -62,14 +61,65 @@ void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) { ...@@ -62,14 +61,65 @@ void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) {
for (int32_t i = 0; i < num; ++i) { for (int32_t i = 0; i < num; ++i) {
SArray *pDb = *(SArray**)taosArrayGet(pResult->pDbVgroup, i); SArray *pDb = *(SArray**)taosArrayGet(pResult->pDbVgroup, i);
int32_t vgNum = taosArrayGetSize(pDb); int32_t vgNum = taosArrayGetSize(pDb);
qDebug("db %d vgInfo:", i);
for (int32_t j = 0; j < vgNum; ++j) { for (int32_t j = 0; j < vgNum; ++j) {
SVgroupInfo* pInfo = taosArrayGet(pDb, j); SVgroupInfo* pInfo = taosArrayGet(pDb, j);
ctgDebug(param, ...); qDebug("vg %d info: vgId:%d", j, pInfo->vgId);
} }
} }
} else { } else {
ctgDebug("empty db vgroup"); qDebug("empty db vgroup");
} }
if (pResult->pTableHash && taosArrayGetSize(pResult->pTableHash) > 0) {
num = taosArrayGetSize(pResult->pTableHash);
for (int32_t i = 0; i < num; ++i) {
SVgroupInfo* pInfo = taosArrayGet(pResult->pTableHash, i);
qDebug("table %d vg info: vgId:%d", i, pInfo->vgId);
}
} else {
qDebug("empty table hash vgroup");
}
if (pResult->pUdfList && taosArrayGetSize(pResult->pUdfList) > 0) {
num = taosArrayGetSize(pResult->pUdfList);
for (int32_t i = 0; i < num; ++i) {
SFuncInfo* pInfo = taosArrayGet(pResult->pUdfList, i);
qDebug("udf %d info: name:%s, funcType:%d", i, pInfo->name, pInfo->funcType);
}
} else {
qDebug("empty udf info");
}
if (pResult->pDbCfg && taosArrayGetSize(pResult->pDbCfg) > 0) {
num = taosArrayGetSize(pResult->pDbCfg);
for (int32_t i = 0; i < num; ++i) {
SDbCfgInfo* pInfo = taosArrayGet(pResult->pDbCfg, i);
qDebug("db %d info: numOFVgroups:%d, numOfStables:%d", i, pInfo->numOfVgroups, pInfo->numOfStables);
}
} else {
qDebug("empty db cfg info");
}
if (pResult->pUser && taosArrayGetSize(pResult->pUser) > 0) {
num = taosArrayGetSize(pResult->pUser);
for (int32_t i = 0; i < num; ++i) {
bool* auth = taosArrayGet(pResult->pUser, i);
qDebug("user auth %d info: %d", i, *auth);
}
} else {
qDebug("empty user auth info");
}
if (pResult->pQnodeList && taosArrayGetSize(pResult->pQnodeList) > 0) {
num = taosArrayGetSize(pResult->pQnodeList);
for (int32_t i = 0; i < num; ++i) {
SQueryNodeAddr* qaddr = taosArrayGet(pResult->pQnodeList, i);
qDebug("qnode %d info: id:%d", i, qaddr->nodeId);
}
} else {
qDebug("empty qnode info");
}
} }
int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId) { int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId) {
......
source/libs/catalog/src/ctgRemote.c
浏览文件 @ 55ec2aa0
...@@ -264,10 +264,11 @@ int32_t ctgGetQnodeListFromMnode(CTG_PARAMS, SArray *out, SCtgTask* pTask) { ...@@ -264,10 +264,11 @@ int32_t ctgGetQnodeListFromMnode(CTG_PARAMS, SArray *out, SCtgTask* pTask) {
char *msg = NULL; char *msg = NULL;
int32_t msgLen = 0; int32_t msgLen = 0;
int32_t reqType = TDMT_MND_QNODE_LIST; int32_t reqType = TDMT_MND_QNODE_LIST;
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get qnode list from mnode, mgmtEpInUse:%d", pMgmtEps->inUse); ctgDebug("try to get qnode list from mnode, mgmtEpInUse:%d", pMgmtEps->inUse);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](NULL, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](NULL, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build qnode list msg failed, error:%s", tstrerror(code)); ctgError("Build qnode list msg failed, error:%s", tstrerror(code));
CTG_ERR_RET(code); CTG_ERR_RET(code);
...@@ -301,10 +302,11 @@ int32_t ctgGetDBVgInfoFromMnode(CTG_PARAMS, SBuildUseDBInput *input, SUseDbOutpu ...@@ -301,10 +302,11 @@ int32_t ctgGetDBVgInfoFromMnode(CTG_PARAMS, SBuildUseDBInput *input, SUseDbOutpu
char *msg = NULL; char *msg = NULL;
int32_t msgLen = 0; int32_t msgLen = 0;
int32_t reqType = TDMT_MND_USE_DB; int32_t reqType = TDMT_MND_USE_DB;
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get db vgInfo from mnode, dbFName:%s", input->db); ctgDebug("try to get db vgInfo from mnode, dbFName:%s", input->db);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](input, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](input, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build use db msg failed, code:%x, db:%s", code, input->db); ctgError("Build use db msg failed, code:%x, db:%s", code, input->db);
CTG_ERR_RET(code); CTG_ERR_RET(code);
...@@ -338,10 +340,11 @@ int32_t ctgGetDBCfgFromMnode(CTG_PARAMS, const char *dbFName, SDbCfgInfo *out, S ...@@ -338,10 +340,11 @@ int32_t ctgGetDBCfgFromMnode(CTG_PARAMS, const char *dbFName, SDbCfgInfo *out, S
char *msg = NULL; char *msg = NULL;
int32_t msgLen = 0; int32_t msgLen = 0;
int32_t reqType = TDMT_MND_GET_DB_CFG; int32_t reqType = TDMT_MND_GET_DB_CFG;
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get db cfg from mnode, dbFName:%s", dbFName); ctgDebug("try to get db cfg from mnode, dbFName:%s", dbFName);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)dbFName, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)dbFName, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build get db cfg msg failed, code:%x, db:%s", code, dbFName); ctgError("Build get db cfg msg failed, code:%x, db:%s", code, dbFName);
CTG_ERR_RET(code); CTG_ERR_RET(code);
...@@ -375,10 +378,11 @@ int32_t ctgGetIndexInfoFromMnode(CTG_PARAMS, const char *indexName, SIndexInfo * ...@@ -375,10 +378,11 @@ int32_t ctgGetIndexInfoFromMnode(CTG_PARAMS, const char *indexName, SIndexInfo *
char *msg = NULL; char *msg = NULL;
int32_t msgLen = 0; int32_t msgLen = 0;
int32_t reqType = TDMT_MND_GET_INDEX; int32_t reqType = TDMT_MND_GET_INDEX;
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get index from mnode, indexName:%s", indexName); ctgDebug("try to get index from mnode, indexName:%s", indexName);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)indexName, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)indexName, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build get index msg failed, code:%x, db:%s", code, indexName); ctgError("Build get index msg failed, code:%x, db:%s", code, indexName);
CTG_ERR_RET(code); CTG_ERR_RET(code);
...@@ -412,10 +416,11 @@ int32_t ctgGetUdfInfoFromMnode(CTG_PARAMS, const char *funcName, SFuncInfo *out, ...@@ -412,10 +416,11 @@ int32_t ctgGetUdfInfoFromMnode(CTG_PARAMS, const char *funcName, SFuncInfo *out,
char *msg = NULL; char *msg = NULL;
int32_t msgLen = 0; int32_t msgLen = 0;
int32_t reqType = TDMT_MND_RETRIEVE_FUNC; int32_t reqType = TDMT_MND_RETRIEVE_FUNC;
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get udf info from mnode, funcName:%s", funcName); ctgDebug("try to get udf info from mnode, funcName:%s", funcName);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)funcName, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)funcName, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build get udf msg failed, code:%x, db:%s", code, funcName); ctgError("Build get udf msg failed, code:%x, db:%s", code, funcName);
CTG_ERR_RET(code); CTG_ERR_RET(code);
...@@ -449,10 +454,11 @@ int32_t ctgGetUserDbAuthFromMnode(CTG_PARAMS, const char *user, SGetUserAuthRsp ...@@ -449,10 +454,11 @@ int32_t ctgGetUserDbAuthFromMnode(CTG_PARAMS, const char *user, SGetUserAuthRsp
char *msg = NULL; char *msg = NULL;
int32_t msgLen = 0; int32_t msgLen = 0;
int32_t reqType = TDMT_MND_GET_USER_AUTH; int32_t reqType = TDMT_MND_GET_USER_AUTH;
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get user auth from mnode, user:%s", user); ctgDebug("try to get user auth from mnode, user:%s", user);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)user, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)]((void *)user, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build get user auth msg failed, code:%x, db:%s", code, user); ctgError("Build get user auth msg failed, code:%x, db:%s", code, user);
CTG_ERR_RET(code); CTG_ERR_RET(code);
...@@ -491,10 +497,11 @@ int32_t ctgGetTbMetaFromMnodeImpl(CTG_PARAMS, char *dbFName, char* tbName, STabl ...@@ -491,10 +497,11 @@ int32_t ctgGetTbMetaFromMnodeImpl(CTG_PARAMS, char *dbFName, char* tbName, STabl
int32_t reqType = TDMT_MND_TABLE_META; int32_t reqType = TDMT_MND_TABLE_META;
char tbFName[TSDB_TABLE_FNAME_LEN]; char tbFName[TSDB_TABLE_FNAME_LEN];
sprintf(tbFName, "%s.%s", dbFName, tbName); sprintf(tbFName, "%s.%s", dbFName, tbName);
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get table meta from mnode, tbFName:%s", tbFName); ctgDebug("try to get table meta from mnode, tbFName:%s", tbFName);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build mnode stablemeta msg failed, code:%x", code); ctgError("Build mnode stablemeta msg failed, code:%x", code);
CTG_ERR_RET(code); CTG_ERR_RET(code);
...@@ -537,6 +544,7 @@ int32_t ctgGetTbMetaFromVnode(CTG_PARAMS, const SName* pTableName, SVgroupInfo * ...@@ -537,6 +544,7 @@ int32_t ctgGetTbMetaFromVnode(CTG_PARAMS, const SName* pTableName, SVgroupInfo *
int32_t reqType = TDMT_VND_TABLE_META; int32_t reqType = TDMT_VND_TABLE_META;
char tbFName[TSDB_TABLE_FNAME_LEN]; char tbFName[TSDB_TABLE_FNAME_LEN];
sprintf(tbFName, "%s.%s", dbFName, pTableName->tname); sprintf(tbFName, "%s.%s", dbFName, pTableName->tname);
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get table meta from vnode, vgId:%d, tbFName:%s", vgroupInfo->vgId, tbFName); ctgDebug("try to get table meta from vnode, vgId:%d, tbFName:%s", vgroupInfo->vgId, tbFName);
...@@ -544,7 +552,7 @@ int32_t ctgGetTbMetaFromVnode(CTG_PARAMS, const SName* pTableName, SVgroupInfo * ...@@ -544,7 +552,7 @@ int32_t ctgGetTbMetaFromVnode(CTG_PARAMS, const SName* pTableName, SVgroupInfo *
char *msg = NULL; char *msg = NULL;
int32_t msgLen = 0; int32_t msgLen = 0;
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen); int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen, mallocFp);
if (code) { if (code) {
ctgError("Build vnode tablemeta msg failed, code:%x, tbFName:%s", code, tbFName); ctgError("Build vnode tablemeta msg failed, code:%x, tbFName:%s", code, tbFName);
CTG_ERR_RET(code); CTG_ERR_RET(code);
......
source/libs/catalog/src/ctgUtil.c
浏览文件 @ 55ec2aa0
...@@ -35,7 +35,7 @@ void ctgFreeSMetaData(SMetaData* pData) { ...@@ -35,7 +35,7 @@ void ctgFreeSMetaData(SMetaData* pData) {
taosArrayDestroy(pData->pUdfList); taosArrayDestroy(pData->pUdfList);
pData->pUdfList = NULL; pData->pUdfList = NULL;
for (int32_t i = 0; i < taosArrayGetSize(pData->pDbCfg); ++i) { for (int32_t i = 0; i < taosArrayGetSize(pData->pDbCfg); ++i) {
SDbCfgInfo* pInfo = taosArrayGet(pData->pDbCfg, i); SDbCfgInfo* pInfo = taosArrayGet(pData->pDbCfg, i);
taosArrayDestroy(pInfo->pRetensions); taosArrayDestroy(pInfo->pRetensions);
...@@ -167,12 +167,15 @@ void ctgFreeHandle(SCatalog* pCtg) { ...@@ -167,12 +167,15 @@ void ctgFreeHandle(SCatalog* pCtg) {
void ctgFreeSUseDbOutput(SUseDbOutput* pOutput) { void ctgFreeSUseDbOutput(SUseDbOutput* pOutput) {
if (NULL == pOutput || NULL == pOutput->dbVgroup) { if (NULL == pOutput) {
return; return;
} }
taosHashCleanup(pOutput->dbVgroup->vgHash); if (pOutput->dbVgroup) {
taosMemoryFreeClear(pOutput->dbVgroup); taosHashCleanup(pOutput->dbVgroup->vgHash);
taosMemoryFreeClear(pOutput->dbVgroup);
}
taosMemoryFree(pOutput); taosMemoryFree(pOutput);
} }
...@@ -267,6 +270,7 @@ void ctgFreeTask(SCtgTask* pTask) { ...@@ -267,6 +270,7 @@ void ctgFreeTask(SCtgTask* pTask) {
switch (pTask->type) { switch (pTask->type) {
case CTG_TASK_GET_QNODE: { case CTG_TASK_GET_QNODE: {
taosArrayDestroy((SArray*)pTask->res); taosArrayDestroy((SArray*)pTask->res);
taosMemoryFreeClear(pTask->taskCtx);
pTask->res = NULL; pTask->res = NULL;
break; break;
} }
...@@ -277,17 +281,19 @@ void ctgFreeTask(SCtgTask* pTask) { ...@@ -277,17 +281,19 @@ void ctgFreeTask(SCtgTask* pTask) {
ctgFreeSTableMetaOutput((STableMetaOutput*)pTask->msgCtx.lastOut); ctgFreeSTableMetaOutput((STableMetaOutput*)pTask->msgCtx.lastOut);
pTask->msgCtx.lastOut = NULL; pTask->msgCtx.lastOut = NULL;
} }
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res); taosMemoryFreeClear(pTask->res);
break; break;
} }
case CTG_TASK_GET_DB_VGROUP: { case CTG_TASK_GET_DB_VGROUP: {
taosArrayDestroy((SArray*)pTask->res); taosArrayDestroy((SArray*)pTask->res);
taosMemoryFreeClear(pTask->taskCtx);
pTask->res = NULL; pTask->res = NULL;
break; break;
} }
case CTG_TASK_GET_DB_CFG: { case CTG_TASK_GET_DB_CFG: {
taosMemoryFreeClear(pTask->taskCtx);
if (pTask->res) { if (pTask->res) {
taosArrayDestroy(((SDbCfgInfo*)pTask->res)->pRetensions);
taosMemoryFreeClear(pTask->res); taosMemoryFreeClear(pTask->res);
} }
break; break;
...@@ -295,6 +301,7 @@ void ctgFreeTask(SCtgTask* pTask) { ...@@ -295,6 +301,7 @@ void ctgFreeTask(SCtgTask* pTask) {
case CTG_TASK_GET_TB_HASH: { case CTG_TASK_GET_TB_HASH: {
SCtgTbHashCtx* taskCtx = (SCtgTbHashCtx*)pTask->taskCtx; SCtgTbHashCtx* taskCtx = (SCtgTbHashCtx*)pTask->taskCtx;
taosMemoryFreeClear(taskCtx->pName); taosMemoryFreeClear(taskCtx->pName);
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res); taosMemoryFreeClear(pTask->res);
break; break;
} }
......
source/libs/parser/src/parTranslater.c
浏览文件 @ 55ec2aa0
...@@ -2766,6 +2766,8 @@ static int32_t buildRollupAst(STranslateContext* pCxt, SCreateTableStmt* pStmt, ...@@ -2766,6 +2766,8 @@ static int32_t buildRollupAst(STranslateContext* pCxt, SCreateTableStmt* pStmt,
break; break;
} }
} }
taosArrayDestroy(dbCfg.pRetensions);
return code; return code;
} }
......
source/libs/qcom/src/querymsg.c
浏览文件 @ 55ec2aa0
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
#pragma GCC diagnostic push #pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wformat-truncation" #pragma GCC diagnostic ignored "-Wformat-truncation"
int32_t (*queryBuildMsg[TDMT_MAX])(void *input, char **msg, int32_t msgSize, int32_t *msgLen) = {0}; int32_t (*queryBuildMsg[TDMT_MAX])(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallocFp)(int32_t)) = {0};
int32_t (*queryProcessMsgRsp[TDMT_MAX])(void *output, char *msg, int32_t msgSize) = {0}; int32_t (*queryProcessMsgRsp[TDMT_MAX])(void *output, char *msg, int32_t msgSize) = {0};
int32_t queryBuildUseDbOutput(SUseDbOutput *pOut, SUseDbRsp *usedbRsp) { int32_t queryBuildUseDbOutput(SUseDbOutput *pOut, SUseDbRsp *usedbRsp) {
...@@ -58,7 +58,7 @@ int32_t queryBuildUseDbOutput(SUseDbOutput *pOut, SUseDbRsp *usedbRsp) { ...@@ -58,7 +58,7 @@ int32_t queryBuildUseDbOutput(SUseDbOutput *pOut, SUseDbRsp *usedbRsp) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
SBuildTableMetaInput *pInput = input; SBuildTableMetaInput *pInput = input;
if (NULL == input || NULL == msg || NULL == msgLen) { if (NULL == input || NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
...@@ -72,7 +72,7 @@ int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int3 ...@@ -72,7 +72,7 @@ int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int3
tstrncpy(infoReq.tbName, pInput->tbName, TSDB_TABLE_NAME_LEN); tstrncpy(infoReq.tbName, pInput->tbName, TSDB_TABLE_NAME_LEN);
int32_t bufLen = tSerializeSTableInfoReq(NULL, 0, &infoReq); int32_t bufLen = tSerializeSTableInfoReq(NULL, 0, &infoReq);
void *pBuf = rpcMallocCont(bufLen); void *pBuf = (*mallcFp)(bufLen);
tSerializeSTableInfoReq(pBuf, bufLen, &infoReq); tSerializeSTableInfoReq(pBuf, bufLen, &infoReq);
*msg = pBuf; *msg = pBuf;
...@@ -81,7 +81,7 @@ int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int3 ...@@ -81,7 +81,7 @@ int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int3
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
SBuildUseDBInput *pInput = input; SBuildUseDBInput *pInput = input;
if (NULL == pInput || NULL == msg || NULL == msgLen) { if (NULL == pInput || NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
...@@ -95,7 +95,7 @@ int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *ms ...@@ -95,7 +95,7 @@ int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *ms
usedbReq.numOfTable = pInput->numOfTable; usedbReq.numOfTable = pInput->numOfTable;
int32_t bufLen = tSerializeSUseDbReq(NULL, 0, &usedbReq); int32_t bufLen = tSerializeSUseDbReq(NULL, 0, &usedbReq);
void *pBuf = rpcMallocCont(bufLen); void *pBuf = (*mallcFp)(bufLen);
tSerializeSUseDbReq(pBuf, bufLen, &usedbReq); tSerializeSUseDbReq(pBuf, bufLen, &usedbReq);
*msg = pBuf; *msg = pBuf;
...@@ -104,7 +104,7 @@ int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *ms ...@@ -104,7 +104,7 @@ int32_t queryBuildUseDbMsg(void *input, char **msg, int32_t msgSize, int32_t *ms
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t queryBuildQnodeListMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { int32_t queryBuildQnodeListMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
if (NULL == msg || NULL == msgLen) { if (NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
} }
...@@ -113,7 +113,7 @@ int32_t queryBuildQnodeListMsg(void *input, char **msg, int32_t msgSize, int32_t ...@@ -113,7 +113,7 @@ int32_t queryBuildQnodeListMsg(void *input, char **msg, int32_t msgSize, int32_t
qnodeListReq.rowNum = -1; qnodeListReq.rowNum = -1;
int32_t bufLen = tSerializeSQnodeListReq(NULL, 0, &qnodeListReq); int32_t bufLen = tSerializeSQnodeListReq(NULL, 0, &qnodeListReq);
void *pBuf = rpcMallocCont(bufLen); void *pBuf = (*mallcFp)(bufLen);
tSerializeSQnodeListReq(pBuf, bufLen, &qnodeListReq); tSerializeSQnodeListReq(pBuf, bufLen, &qnodeListReq);
*msg = pBuf; *msg = pBuf;
...@@ -122,7 +122,7 @@ int32_t queryBuildQnodeListMsg(void *input, char **msg, int32_t msgSize, int32_t ...@@ -122,7 +122,7 @@ int32_t queryBuildQnodeListMsg(void *input, char **msg, int32_t msgSize, int32_t
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
if (NULL == msg || NULL == msgLen) { if (NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
} }
...@@ -131,7 +131,7 @@ int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t ...@@ -131,7 +131,7 @@ int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t
strcpy(dbCfgReq.db, input); strcpy(dbCfgReq.db, input);
int32_t bufLen = tSerializeSDbCfgReq(NULL, 0, &dbCfgReq); int32_t bufLen = tSerializeSDbCfgReq(NULL, 0, &dbCfgReq);
void *pBuf = rpcMallocCont(bufLen); void *pBuf = (*mallcFp)(bufLen);
tSerializeSDbCfgReq(pBuf, bufLen, &dbCfgReq); tSerializeSDbCfgReq(pBuf, bufLen, &dbCfgReq);
*msg = pBuf; *msg = pBuf;
...@@ -140,7 +140,7 @@ int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t ...@@ -140,7 +140,7 @@ int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t queryBuildGetIndexMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { int32_t queryBuildGetIndexMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
if (NULL == msg || NULL == msgLen) { if (NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
} }
...@@ -149,7 +149,7 @@ int32_t queryBuildGetIndexMsg(void *input, char **msg, int32_t msgSize, int32_t ...@@ -149,7 +149,7 @@ int32_t queryBuildGetIndexMsg(void *input, char **msg, int32_t msgSize, int32_t
strcpy(indexReq.indexFName, input); strcpy(indexReq.indexFName, input);
int32_t bufLen = tSerializeSUserIndexReq(NULL, 0, &indexReq); int32_t bufLen = tSerializeSUserIndexReq(NULL, 0, &indexReq);
void *pBuf = rpcMallocCont(bufLen); void *pBuf = (*mallcFp)(bufLen);
tSerializeSUserIndexReq(pBuf, bufLen, &indexReq); tSerializeSUserIndexReq(pBuf, bufLen, &indexReq);
*msg = pBuf; *msg = pBuf;
...@@ -158,7 +158,7 @@ int32_t queryBuildGetIndexMsg(void *input, char **msg, int32_t msgSize, int32_t ...@@ -158,7 +158,7 @@ int32_t queryBuildGetIndexMsg(void *input, char **msg, int32_t msgSize, int32_t
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
if (NULL == msg || NULL == msgLen) { if (NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
} }
...@@ -170,7 +170,7 @@ int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int3 ...@@ -170,7 +170,7 @@ int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int3
taosArrayPush(funcReq.pFuncNames, input); taosArrayPush(funcReq.pFuncNames, input);
int32_t bufLen = tSerializeSRetrieveFuncReq(NULL, 0, &funcReq); int32_t bufLen = tSerializeSRetrieveFuncReq(NULL, 0, &funcReq);
void *pBuf = rpcMallocCont(bufLen); void *pBuf = (*mallcFp)(bufLen);
tSerializeSRetrieveFuncReq(pBuf, bufLen, &funcReq); tSerializeSRetrieveFuncReq(pBuf, bufLen, &funcReq);
taosArrayDestroy(funcReq.pFuncNames); taosArrayDestroy(funcReq.pFuncNames);
...@@ -181,7 +181,7 @@ int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int3 ...@@ -181,7 +181,7 @@ int32_t queryBuildRetrieveFuncMsg(void *input, char **msg, int32_t msgSize, int3
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t queryBuildGetUserAuthMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen) { int32_t queryBuildGetUserAuthMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
if (NULL == msg || NULL == msgLen) { if (NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT; return TSDB_CODE_TSC_INVALID_INPUT;
} }
...@@ -190,7 +190,7 @@ int32_t queryBuildGetUserAuthMsg(void *input, char **msg, int32_t msgSize, int32 ...@@ -190,7 +190,7 @@ int32_t queryBuildGetUserAuthMsg(void *input, char **msg, int32_t msgSize, int32
strncpy(req.user, input, sizeof(req.user)); strncpy(req.user, input, sizeof(req.user));
int32_t bufLen = tSerializeSGetUserAuthReq(NULL, 0, &req); int32_t bufLen = tSerializeSGetUserAuthReq(NULL, 0, &req);
void *pBuf = rpcMallocCont(bufLen); void *pBuf = (*mallcFp)(bufLen);
tSerializeSGetUserAuthReq(pBuf, bufLen, &req); tSerializeSGetUserAuthReq(pBuf, bufLen, &req);
*msg = pBuf; *msg = pBuf;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或