enh: add catalog return code processing
This commit is contained in:
dapan1121
parent
5cd03bb2ce
commit
ecc2ae7757
|
|
@ -68,7 +68,7 @@ int32_t ctgInitGetTbMetaTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
|
|||
TAOS_MEMCPY(ctx->pName, name, sizeof(*name));
|
||||
ctx->flag = pParam->flag | CTG_FLAG_UNKNOWN_STB;
|
||||
|
||||
if (NULL != taosArrayPush(pJob->pTasks, &task)) {
|
||||
if (NULL == taosArrayPush(pJob->pTasks, &task)) {
|
||||
ctgFreeTask(&task, true);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -57,8 +57,23 @@ int32_t ctgHandleBatchRsp(SCtgJob* pJob, SCtgTaskCallbackParam* cbParam, SDataBu
|
|||
|
||||
for (int32_t i = 0; i < taskNum; ++i) {
|
||||
int32_t* taskId = taosArrayGet(cbParam->taskId, i);
|
||||
if (NULL == taskId) {
|
||||
ctgError("taosArrayGet %d taskId failed, total:%d", i, (int32_t)taosArrayGetSize(cbParam->taskId));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
int32_t* msgIdx = taosArrayGet(cbParam->msgIdx, i);
|
||||
if (NULL == msgIdx) {
|
||||
ctgError("taosArrayGet %d msgIdx failed, total:%d", i, (int32_t)taosArrayGetSize(cbParam->msgIdx));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
SCtgTask* pTask = taosArrayGet(pJob->pTasks, *taskId);
|
||||
if (NULL == pTask) {
|
||||
ctgError("taosArrayGet %d SCtgTask failed, total:%d", *taskId, (int32_t)taosArrayGetSize(pJob->pTasks));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
if (msgNum > 0) {
|
||||
pRsp = taosArrayGet(batchRsp.pRsps, i);
|
||||
|
||||
|
|
@ -89,12 +104,17 @@ int32_t ctgHandleBatchRsp(SCtgJob* pJob, SCtgTaskCallbackParam* cbParam, SDataBu
|
|||
tReq.pTask = pTask;
|
||||
tReq.msgIdx = pRsp->msgIdx;
|
||||
SCtgMsgCtx* pMsgCtx = CTG_GET_TASK_MSGCTX(pTask, tReq.msgIdx);
|
||||
if (NULL == pMsgCtx) {
|
||||
ctgError("get task %d SCtgMsgCtx failed, taskType:%d", tReq.msgIdx, pTask->type);
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
pMsgCtx->pBatchs = pBatchs;
|
||||
|
||||
ctgDebug("QID:0x%" PRIx64 " ctg task %d idx %d start to handle rsp %s, pBatchs: %p", pJob->queryId, pTask->taskId,
|
||||
pRsp->msgIdx, TMSG_INFO(taskMsg.msgType + 1), pBatchs);
|
||||
|
||||
(*gCtgAsyncFps[pTask->type].handleRspFp)(&tReq, pRsp->reqType, &taskMsg, (pRsp->rspCode ? pRsp->rspCode : rspCode));
|
||||
(void)(*gCtgAsyncFps[pTask->type].handleRspFp)(&tReq, pRsp->reqType, &taskMsg, (pRsp->rspCode ? pRsp->rspCode : rspCode)); // error handled internal
|
||||
}
|
||||
|
||||
CTG_ERR_JRET(ctgLaunchBatchs(pJob->pCtg, pJob, pBatchs));
|
||||
|
|
@ -398,7 +418,16 @@ int32_t ctgHandleMsgCallback(void* param, SDataBuf* pMsg, int32_t rspCode) {
|
|||
CTG_ERR_JRET(ctgHandleBatchRsp(pJob, cbParam, pMsg, rspCode));
|
||||
} else {
|
||||
int32_t* taskId = taosArrayGet(cbParam->taskId, 0);
|
||||
if (NULL == taskId) {
|
||||
ctgError("taosArrayGet %d taskId failed, total:%d", 0, (int32_t)taosArrayGetSize(cbParam->taskId));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
SCtgTask* pTask = taosArrayGet(pJob->pTasks, *taskId);
|
||||
if (NULL == pTask) {
|
||||
ctgError("taosArrayGet %d SCtgTask failed, total:%d", *taskId, (int32_t)taosArrayGetSize(pJob->pTasks));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
qDebug("QID:0x%" PRIx64 " ctg task %d start to handle rsp %s", pJob->queryId, pTask->taskId,
|
||||
TMSG_INFO(cbParam->reqType + 1));
|
||||
|
|
@ -412,6 +441,11 @@ int32_t ctgHandleMsgCallback(void* param, SDataBuf* pMsg, int32_t rspCode) {
|
|||
}
|
||||
|
||||
SCtgMsgCtx* pMsgCtx = CTG_GET_TASK_MSGCTX(pTask, -1);
|
||||
if (NULL == pMsgCtx) {
|
||||
ctgError("get task %d SCtgMsgCtx failed, taskType:%d", -1, pTask->type);
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
pMsgCtx->pBatchs = pBatchs;
|
||||
#endif
|
||||
|
||||
|
|
@ -432,7 +466,7 @@ _return:
|
|||
taosMemoryFree(pMsg->pEpSet);
|
||||
|
||||
if (pJob) {
|
||||
taosReleaseRef(gCtgMgmt.jobPool, cbParam->refId);
|
||||
(void)taosReleaseRef(gCtgMgmt.jobPool, cbParam->refId);
|
||||
}
|
||||
|
||||
CTG_API_LEAVE(code);
|
||||
|
|
@ -450,6 +484,7 @@ int32_t ctgMakeMsgSendInfo(SCtgJob* pJob, SArray* pTaskId, int32_t batchId, SArr
|
|||
SCtgTaskCallbackParam* param = taosMemoryCalloc(1, sizeof(SCtgTaskCallbackParam));
|
||||
if (NULL == param) {
|
||||
qError("calloc %d failed", (int32_t)sizeof(SCtgTaskCallbackParam));
|
||||
taosMemoryFree(msgSendInfo);
|
||||
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
|
|
@ -482,7 +517,7 @@ int32_t ctgAsyncSendMsg(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob* pJob,
|
|||
SMsgSendInfo* pMsgSendInfo = NULL;
|
||||
CTG_ERR_JRET(ctgMakeMsgSendInfo(pJob, pTaskId, batchId, pMsgIdx, msgType, &pMsgSendInfo));
|
||||
|
||||
ctgUpdateSendTargetInfo(pMsgSendInfo, msgType, dbFName, vgId);
|
||||
CTG_ERR_JRET(ctgUpdateSendTargetInfo(pMsgSendInfo, msgType, dbFName, vgId));
|
||||
|
||||
pMsgSendInfo->requestId = pConn->requestId;
|
||||
pMsgSendInfo->requestObjRefId = pConn->requestObjRefId;
|
||||
|
|
@ -515,18 +550,25 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
|
|||
void* msg, uint32_t msgSize) {
|
||||
int32_t code = 0;
|
||||
SCtgTask* pTask = tReq->pTask;
|
||||
SCtgMsgCtx* pMsgCtx = CTG_GET_TASK_MSGCTX(pTask, tReq->msgIdx);
|
||||
SHashObj* pBatchs = pMsgCtx->pBatchs;
|
||||
SCtgJob* pJob = pTask->pJob;
|
||||
SCtgBatch* pBatch = taosHashGet(pBatchs, &vgId, sizeof(vgId));
|
||||
SCtgBatch newBatch = {0};
|
||||
SBatchMsg req = {0};
|
||||
SCtgMsgCtx* pMsgCtx = CTG_GET_TASK_MSGCTX(pTask, tReq->msgIdx);
|
||||
if (NULL == pMsgCtx) {
|
||||
ctgError("get task %d SCtgMsgCtx failed, taskType:%d", tReq->msgIdx, pTask->type);
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
SHashObj* pBatchs = pMsgCtx->pBatchs;
|
||||
SCtgBatch* pBatch = taosHashGet(pBatchs, &vgId, sizeof(vgId));
|
||||
if (NULL == pBatch) {
|
||||
newBatch.pMsgs = taosArrayInit(pJob->subTaskNum, sizeof(SBatchMsg));
|
||||
newBatch.pTaskIds = taosArrayInit(pJob->subTaskNum, sizeof(int32_t));
|
||||
newBatch.pMsgIdxs = taosArrayInit(pJob->subTaskNum, sizeof(int32_t));
|
||||
if (NULL == newBatch.pMsgs || NULL == newBatch.pTaskIds || NULL == newBatch.pMsgIdxs) {
|
||||
taosArrayDestroy(newBatch.pMsgs);
|
||||
taosArrayDestroy(newBatch.pTaskIds);
|
||||
taosArrayDestroy(newBatch.pMsgIdxs);
|
||||
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
|
|
@ -562,6 +604,10 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
|
|||
SCtgTSMAFetch* pFetch = taosArrayGet(pCtx->pFetches, tReq->msgIdx);
|
||||
STablesReq* pTbReq = taosArrayGet(pCtx->pNames, pFetch->dbIdx);
|
||||
pName = taosArrayGet(pTbReq->pTables, pFetch->tbIdx);
|
||||
if (NULL == pName) {
|
||||
ctgError("fail to get %d SName, totalTables:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pTbReq->pTables));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
} else {
|
||||
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
|
||||
pName = ctx->pName;
|
||||
|
|
@ -569,14 +615,26 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
|
|||
} else if (TDMT_VND_GET_STREAM_PROGRESS == msgType) {
|
||||
SCtgTbTSMACtx* pCtx = pTask->taskCtx;
|
||||
SCtgTSMAFetch* pFetch = taosArrayGet(pCtx->pFetches, tReq->msgIdx);
|
||||
if (NULL == pFetch) {
|
||||
ctgError("fail to get %d SCtgTSMAFetch, totalFetchs:%d", tReq->msgIdx, (int32_t)taosArrayGetSize(pCtx->pFetches));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
STablesReq* pTbReq = taosArrayGet(pCtx->pNames, pFetch->dbIdx);
|
||||
if (NULL == pTbReq) {
|
||||
ctgError("fail to get %d STablesReq, totalTables:%d", pFetch->dbIdx, (int32_t)taosArrayGetSize(pCtx->pNames));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
pName = taosArrayGet(pTbReq->pTables, pFetch->tbIdx);
|
||||
if (NULL == pName) {
|
||||
ctgError("fail to get %d SName, totalTables:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pTbReq->pTables));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
} else {
|
||||
ctgError("invalid vnode msgType %d", msgType);
|
||||
CTG_ERR_JRET(TSDB_CODE_APP_ERROR);
|
||||
}
|
||||
|
||||
tNameGetFullDbName(pName, newBatch.dbFName);
|
||||
(void)tNameGetFullDbName(pName, newBatch.dbFName);
|
||||
}
|
||||
|
||||
newBatch.msgType = (vgId > 0) ? TDMT_VND_BATCH_META : TDMT_MND_BATCH_META;
|
||||
|
|
@ -622,6 +680,10 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
|
|||
SCtgTSMAFetch* pFetch = taosArrayGet(pCtx->pFetches, tReq->msgIdx);
|
||||
STablesReq* pTbReq = taosArrayGet(pCtx->pNames, pFetch->dbIdx);
|
||||
pName = taosArrayGet(pTbReq->pTables, pFetch->tbIdx);
|
||||
if (NULL == pName) {
|
||||
ctgError("fail to get %d SName, totalTables:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pTbReq->pTables));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
} else {
|
||||
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
|
||||
pName = ctx->pName;
|
||||
|
|
@ -629,14 +691,26 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
|
|||
} else if (TDMT_VND_GET_STREAM_PROGRESS == msgType) {
|
||||
SCtgTbTSMACtx* pCtx = pTask->taskCtx;
|
||||
SCtgTSMAFetch* pFetch = taosArrayGet(pCtx->pFetches, tReq->msgIdx);
|
||||
if (NULL == pFetch) {
|
||||
ctgError("fail to get %d SCtgTSMAFetch, totalFetchs:%d", tReq->msgIdx, (int32_t)taosArrayGetSize(pCtx->pFetches));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
STablesReq* pTbReq = taosArrayGet(pCtx->pNames, pFetch->dbIdx);
|
||||
if (NULL == pTbReq) {
|
||||
ctgError("fail to get %d STablesReq, totalTables:%d", pFetch->dbIdx, (int32_t)taosArrayGetSize(pCtx->pNames));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
pName = taosArrayGet(pTbReq->pTables, pFetch->tbIdx);
|
||||
if (NULL == pName) {
|
||||
ctgError("fail to get %d SName, totalTables:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pTbReq->pTables));
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
} else {
|
||||
ctgError("invalid vnode msgType %d", msgType);
|
||||
CTG_ERR_JRET(TSDB_CODE_APP_ERROR);
|
||||
}
|
||||
|
||||
tNameGetFullDbName(pName, pBatch->dbFName);
|
||||
(void)tNameGetFullDbName(pName, pBatch->dbFName);
|
||||
}
|
||||
|
||||
ctgDebug("task %d %s req added to batch %d, target vgId %d", pTask->taskId, TMSG_INFO(msgType), pBatch->batchId,
|
||||
|
|
@ -752,7 +826,10 @@ int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SArray
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -765,7 +842,7 @@ int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SArray
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, NULL));
|
||||
|
||||
|
|
@ -801,7 +878,10 @@ int32_t ctgGetDnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SArray
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -814,7 +894,7 @@ int32_t ctgGetDnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SArray
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, NULL));
|
||||
|
||||
|
|
@ -854,7 +934,10 @@ int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SBuildU
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -867,7 +950,7 @@ int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SBuildU
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, input->db));
|
||||
|
||||
|
|
@ -909,7 +992,10 @@ int32_t ctgGetDBCfgFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const char
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -922,7 +1008,7 @@ int32_t ctgGetDBCfgFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const char
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)dbFName));
|
||||
|
||||
|
|
@ -964,7 +1050,10 @@ int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -977,7 +1066,7 @@ int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)indexName));
|
||||
|
||||
|
|
@ -993,7 +1082,7 @@ int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName* n
|
|||
int32_t reqType = TDMT_MND_GET_TABLE_INDEX;
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(name, tbFName);
|
||||
(void)tNameExtractFullName(name, tbFName);
|
||||
|
||||
ctgDebug("try to get tb index from mnode, tbFName:%s", tbFName);
|
||||
|
||||
|
|
@ -1021,7 +1110,10 @@ int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName* n
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1034,7 +1126,7 @@ int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName* n
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)tbFName));
|
||||
|
||||
|
|
@ -1076,7 +1168,10 @@ int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const ch
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1089,7 +1184,7 @@ int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const ch
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)funcName));
|
||||
|
||||
|
|
@ -1131,7 +1226,10 @@ int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1144,7 +1242,7 @@ int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)user));
|
||||
|
||||
|
|
@ -1162,7 +1260,7 @@ int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
int32_t msgLen = 0;
|
||||
int32_t reqType = TDMT_MND_TABLE_META;
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
sprintf(tbFName, "%s.%s", dbFName, tbName);
|
||||
(void)sprintf(tbFName, "%s.%s", dbFName, tbName);
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
|
||||
ctgDebug("try to get table meta from mnode, tbFName:%s", tbFName);
|
||||
|
|
@ -1188,7 +1286,10 @@ int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1201,7 +1302,7 @@ int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, tbFName));
|
||||
|
||||
|
|
@ -1213,7 +1314,7 @@ int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo* pConn, const
|
|||
int32_t ctgGetTbMetaFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, STableMetaOutput* out,
|
||||
SCtgTaskReq* tReq) {
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(pTableName, dbFName);
|
||||
(void)tNameGetFullDbName(pTableName, dbFName);
|
||||
|
||||
return ctgGetTbMetaFromMnodeImpl(pCtg, pConn, dbFName, (char*)pTableName->tname, out, tReq);
|
||||
}
|
||||
|
|
@ -1222,10 +1323,10 @@ int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
|
|||
STableMetaOutput* out, SCtgTaskReq* tReq) {
|
||||
SCtgTask* pTask = tReq ? tReq->pTask : NULL;
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(pTableName, dbFName);
|
||||
(void)tNameGetFullDbName(pTableName, dbFName);
|
||||
int32_t reqType = TDMT_VND_TABLE_META;
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
sprintf(tbFName, "%s.%s", dbFName, pTableName->tname);
|
||||
(void)sprintf(tbFName, "%s.%s", dbFName, pTableName->tname);
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
|
||||
SEp* pEp = &vgroupInfo->epSet.eps[vgroupInfo->epSet.inUse];
|
||||
|
|
@ -1261,12 +1362,15 @@ int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
|
|||
#else
|
||||
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(ctx->pName, dbFName);
|
||||
(void)tNameGetFullDbName(ctx->pName, dbFName);
|
||||
SArray* pTaskId = taosArrayInit(1, sizeof(int32_t));
|
||||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, &vConn, pTask->pJob, pTaskId, -1, NULL, dbFName, ctx->vgId, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1279,7 +1383,7 @@ int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, tbFName));
|
||||
|
||||
|
|
@ -1294,10 +1398,10 @@ int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
|
|||
int32_t msgLen = 0;
|
||||
int32_t reqType = TDMT_VND_TABLE_CFG;
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(pTableName, tbFName);
|
||||
(void)tNameExtractFullName(pTableName, tbFName);
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(pTableName, dbFName);
|
||||
(void)tNameGetFullDbName(pTableName, dbFName);
|
||||
SBuildTableInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char*)pTableName->tname};
|
||||
|
||||
SEp* pEp = &vgroupInfo->epSet.eps[vgroupInfo->epSet.inUse];
|
||||
|
|
@ -1325,12 +1429,15 @@ int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
|
|||
#else
|
||||
SCtgTbCfgCtx* ctx = (SCtgTbCfgCtx*)pTask->taskCtx;
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(ctx->pName, dbFName);
|
||||
(void)tNameGetFullDbName(ctx->pName, dbFName);
|
||||
SArray* pTaskId = taosArrayInit(1, sizeof(int32_t));
|
||||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, &vConn, pTask->pJob, pTaskId, -1, NULL, dbFName, ctx->pVgInfo->vgId, reqType, msg,
|
||||
msgLen));
|
||||
|
|
@ -1344,7 +1451,7 @@ int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)tbFName));
|
||||
|
||||
|
|
@ -1359,10 +1466,10 @@ int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
|
|||
int32_t msgLen = 0;
|
||||
int32_t reqType = TDMT_MND_TABLE_CFG;
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(pTableName, tbFName);
|
||||
(void)tNameExtractFullName(pTableName, tbFName);
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(pTableName, dbFName);
|
||||
(void)tNameGetFullDbName(pTableName, dbFName);
|
||||
SBuildTableInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = (char*)pTableName->tname};
|
||||
|
||||
ctgDebug("try to get table cfg from mnode, tbFName:%s", tbFName);
|
||||
|
|
@ -1386,7 +1493,10 @@ int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1399,7 +1509,7 @@ int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)tbFName));
|
||||
|
||||
|
|
@ -1435,7 +1545,10 @@ int32_t ctgGetSvrVerFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, char** ou
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1448,7 +1561,7 @@ int32_t ctgGetSvrVerFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, char** ou
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, NULL));
|
||||
|
||||
|
|
@ -1465,7 +1578,7 @@ int32_t ctgGetViewInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName*
|
|||
SCtgTask* pTask = tReq ? tReq->pTask : NULL;
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
char fullName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(pName, fullName);
|
||||
(void)tNameExtractFullName(pName, fullName);
|
||||
|
||||
ctgDebug("try to get view info from mnode, viewFName:%s", fullName);
|
||||
|
||||
|
|
@ -1490,7 +1603,10 @@ int32_t ctgGetViewInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName*
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1503,7 +1619,7 @@ int32_t ctgGetViewInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName*
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, fullName));
|
||||
|
||||
|
|
@ -1519,7 +1635,7 @@ int32_t ctgGetTbTSMAFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
|
|||
SCtgTask* pTask = tReq ? tReq->pTask : NULL;
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(name, tbFName);
|
||||
(void)tNameExtractFullName(name, tbFName);
|
||||
|
||||
ctgDebug("try to get tb index from mnode, tbFName:%s", tbFName);
|
||||
|
||||
|
|
@ -1544,7 +1660,10 @@ int32_t ctgGetTbTSMAFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
|
|||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask->pJob, pTaskId, -1, NULL, NULL, 0, reqType, msg, msgLen));
|
||||
#endif
|
||||
|
|
@ -1557,7 +1676,7 @@ int32_t ctgGetTbTSMAFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)tbFName));
|
||||
|
||||
|
|
@ -1573,7 +1692,7 @@ int32_t ctgGetStreamProgressFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, c
|
|||
int32_t msgLen = 0;
|
||||
int32_t reqType = TDMT_VND_GET_STREAM_PROGRESS;
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(pTbName, tbFName);
|
||||
(void)tNameExtractFullName(pTbName, tbFName);
|
||||
SCtgTask* pTask = tReq ? tReq->pTask : NULL;
|
||||
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
|
||||
|
||||
|
|
@ -1602,15 +1721,17 @@ int32_t ctgGetStreamProgressFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, c
|
|||
CTG_RET(ctgAddBatch(pCtg, vgroupInfo->vgId, &vConn, tReq, reqType, msg, msgLen));
|
||||
#else
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(pTbName, dbFName);
|
||||
(void)tNameGetFullDbName(pTbName, dbFName);
|
||||
SArray* pTaskId = taosArrayInit(1, sizeof(int32_t));
|
||||
if (NULL == pTaskId) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
taosArrayPush(pTaskId, &pTask->taskId);
|
||||
if (NULL == taosArrayPush(pTaskId, &pTask->taskId)) {
|
||||
taosArrayDestroy(pTaskId);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
CTG_RET(
|
||||
ctgAsyncSendMsg(pCtg, &vConn, pTask->pJob, pTaskId, -1, NULL, dbFName, vgroupInfo->vgId, reqType, msg, msgLen));
|
||||
CTG_RET(ctgAsyncSendMsg(pCtg, &vConn, pTask->pJob, pTaskId, -1, NULL, dbFName, vgroupInfo->vgId, reqType, msg, msgLen));
|
||||
#endif
|
||||
}
|
||||
|
||||
|
|
@ -1621,7 +1742,7 @@ int32_t ctgGetStreamProgressFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, c
|
|||
};
|
||||
|
||||
SRpcMsg rpcRsp = {0};
|
||||
rpcSendRecv(pConn->pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp);
|
||||
CTG_ERR_RET(rpcSendRecv(pConn->pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp));
|
||||
|
||||
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)tbFName));
|
||||
|
||||
|
|
|
|||
|
|
@ -100,7 +100,7 @@ int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t si
|
|||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
memcpy(orig, meta, size);
|
||||
TAOS_MEMCPY(orig, meta, size);
|
||||
|
||||
qDebug("meta in rent updated, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
|
||||
|
||||
|
|
@ -185,8 +185,12 @@ int32_t ctgMetaRentGetImpl(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_
|
|||
}
|
||||
|
||||
void *meta = taosArrayGet(slot->meta, 0);
|
||||
if (NULL == meta) {
|
||||
qError("get the 0th meta in slot failed, total:%d", (int32_t)metaNum);
|
||||
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
memcpy(*res, meta, msize);
|
||||
TAOS_MEMCPY(*res, meta, msize);
|
||||
|
||||
*num = (uint32_t)metaNum;
|
||||
|
||||
|
|
@ -227,13 +231,14 @@ void ctgRemoveStbRent(SCatalog *pCtg, SCtgDBCache *dbCache) {
|
|||
return;
|
||||
}
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
void *pIter = taosHashIterate(dbCache->stbCache, NULL);
|
||||
while (pIter) {
|
||||
uint64_t *suid = NULL;
|
||||
suid = taosHashGetKey(pIter, NULL);
|
||||
|
||||
if (TSDB_CODE_SUCCESS ==
|
||||
ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare)) {
|
||||
code = ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare);
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
ctgDebug("stb removed from rent, suid:0x%" PRIx64, *suid);
|
||||
}
|
||||
|
||||
|
|
@ -265,6 +270,7 @@ void ctgRemoveTSMARent(SCatalog *pCtg, SCtgDBCache *dbCache) {
|
|||
void* pIter = taosHashIterate(dbCache->tsmaCache, NULL);
|
||||
while (pIter) {
|
||||
SCtgTSMACache* pCtgCache = pIter;
|
||||
|
||||
CTG_LOCK(CTG_READ, &pCtgCache->tsmaLock);
|
||||
int32_t size = (pCtgCache && pCtgCache->pTsmas) ? pCtgCache->pTsmas->size : 0;
|
||||
for (int32_t i = 0; i < size; ++i) {
|
||||
|
|
@ -274,6 +280,7 @@ void ctgRemoveTSMARent(SCatalog *pCtg, SCtgDBCache *dbCache) {
|
|||
}
|
||||
}
|
||||
CTG_UNLOCK(CTG_READ, &pCtgCache->tsmaLock);
|
||||
|
||||
pIter = taosHashIterate(dbCache->tsmaCache, pIter);
|
||||
}
|
||||
}
|
||||
|
|
@ -325,8 +332,10 @@ int32_t ctgUpdateRentTSMAVersion(SCatalog *pCtg, char *dbFName, const STSMACache
|
|||
tstrncpy(tsmaRent.name, pTsmaInfo->name, TSDB_TABLE_NAME_LEN);
|
||||
tstrncpy(tsmaRent.dbFName, dbFName, TSDB_DB_FNAME_LEN);
|
||||
tstrncpy(tsmaRent.tbName, pTsmaInfo->tb, TSDB_TABLE_NAME_LEN);
|
||||
|
||||
CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->tsmaRent, &tsmaRent, tsmaRent.tsmaId, sizeof(STSMAVersion),
|
||||
ctgTSMAVersionSortCompare, ctgTSMAVersionSearchCompare));
|
||||
|
||||
ctgDebug("db %s, 0x%" PRIx64 " tsma %s, 0x%" PRIx64 "version %d updated to tsmaRent", dbFName, tsmaRent.dbId,
|
||||
pTsmaInfo->name, pTsmaInfo->tsmaId, pTsmaInfo->version);
|
||||
|
||||
|
|
|
|||
|
|
@ -107,6 +107,10 @@ char* ctgTaskTypeStr(CTG_TASK_TYPE type) {
|
|||
return "[get table tag]";
|
||||
case CTG_TASK_GET_VIEW:
|
||||
return "[get view]";
|
||||
case CTG_TASK_GET_TB_TSMA:
|
||||
return "[get table TSMA]";
|
||||
case CTG_TASK_GET_TSMA:
|
||||
return "[get TSMA]";
|
||||
default:
|
||||
return "unknown";
|
||||
}
|
||||
|
|
@ -443,10 +447,10 @@ void ctgClearHandleMeta(SCatalog* pCtg, int64_t *pClearedSize, int64_t *pCleardN
|
|||
continue;
|
||||
}
|
||||
|
||||
taosHashRemove(dbCache->tbCache, key, len);
|
||||
cacheSize =
|
||||
len + sizeof(SCtgTbCache) + ctgGetTbMetaCacheSize(pCache->pMeta) + ctgGetTbIndexCacheSize(pCache->pIndex);
|
||||
atomic_sub_fetch_64(&dbCache->dbCacheSize, cacheSize);
|
||||
(void)taosHashRemove(dbCache->tbCache, key, len);
|
||||
|
||||
cacheSize = len + sizeof(SCtgTbCache) + ctgGetTbMetaCacheSize(pCache->pMeta) + ctgGetTbIndexCacheSize(pCache->pIndex);
|
||||
(void)atomic_sub_fetch_64(&dbCache->dbCacheSize, cacheSize);
|
||||
*pClearedSize += cacheSize;
|
||||
(*pCleardNum)++;
|
||||
|
||||
|
|
@ -508,10 +512,10 @@ void ctgClearHandle(SCatalog* pCtg) {
|
|||
ctgFreeInstDbCache(pCtg->dbCache);
|
||||
ctgFreeInstUserCache(pCtg->userCache);
|
||||
|
||||
ctgMetaRentInit(&pCtg->dbRent, gCtgMgmt.cfg.dbRentSec, CTG_RENT_DB, sizeof(SDbCacheInfo));
|
||||
ctgMetaRentInit(&pCtg->stbRent, gCtgMgmt.cfg.stbRentSec, CTG_RENT_STABLE, sizeof(SSTableVersion));
|
||||
ctgMetaRentInit(&pCtg->viewRent, gCtgMgmt.cfg.viewRentSec, CTG_RENT_VIEW, sizeof(SViewVersion));
|
||||
ctgMetaRentInit(&pCtg->tsmaRent, gCtgMgmt.cfg.tsmaRentSec, CTG_RENT_TSMA, sizeof(STSMAVersion));
|
||||
(void)ctgMetaRentInit(&pCtg->dbRent, gCtgMgmt.cfg.dbRentSec, CTG_RENT_DB, sizeof(SDbCacheInfo));
|
||||
(void)ctgMetaRentInit(&pCtg->stbRent, gCtgMgmt.cfg.stbRentSec, CTG_RENT_STABLE, sizeof(SSTableVersion));
|
||||
(void)ctgMetaRentInit(&pCtg->viewRent, gCtgMgmt.cfg.viewRentSec, CTG_RENT_VIEW, sizeof(SViewVersion));
|
||||
(void)ctgMetaRentInit(&pCtg->tsmaRent, gCtgMgmt.cfg.tsmaRentSec, CTG_RENT_TSMA, sizeof(STSMAVersion));
|
||||
|
||||
pCtg->dbCache = taosHashInit(gCtgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false,
|
||||
HASH_ENTRY_LOCK);
|
||||
|
|
@ -525,7 +529,7 @@ void ctgClearHandle(SCatalog* pCtg) {
|
|||
ctgError("taosHashInit %d user cache failed", gCtgMgmt.cfg.maxUserCacheNum);
|
||||
}
|
||||
|
||||
memset(pCtg->cacheStat.cacheNum, 0, sizeof(pCtg->cacheStat.cacheNum));
|
||||
TAOS_MEMSET(pCtg->cacheStat.cacheNum, 0, sizeof(pCtg->cacheStat.cacheNum));
|
||||
|
||||
CTG_STAT_RT_INC(numOfOpClearCache, 1);
|
||||
|
||||
|
|
@ -668,7 +672,7 @@ void ctgFreeSTableMetaOutput(STableMetaOutput* pOutput) {
|
|||
|
||||
void ctgResetTbMetaTask(SCtgTask* pTask) {
|
||||
SCtgTbMetaCtx* taskCtx = (SCtgTbMetaCtx*)pTask->taskCtx;
|
||||
memset(&taskCtx->tbInfo, 0, sizeof(taskCtx->tbInfo));
|
||||
TAOS_MEMSET(&taskCtx->tbInfo, 0, sizeof(taskCtx->tbInfo));
|
||||
taskCtx->flag = CTG_FLAG_UNKNOWN_STB;
|
||||
|
||||
if (pTask->msgCtx.lastOut) {
|
||||
|
|
@ -1087,7 +1091,10 @@ int32_t ctgAddMsgCtx(SArray* pCtxs, int32_t reqType, void* out, char* target) {
|
|||
}
|
||||
}
|
||||
|
||||
taosArrayPush(pCtxs, &ctx);
|
||||
if (NULL == taosArrayPush(pCtxs, &ctx)) {
|
||||
ctgFreeMsgCtx(&ctx);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
|
@ -1174,13 +1181,14 @@ int32_t ctgGetVgInfoFromHashValue(SCatalog* pCtg, SEpSet* pMgmtEps, SDBVgInfo* d
|
|||
|
||||
int32_t vgNum = taosArrayGetSize(dbInfo->vgArray);
|
||||
char db[TSDB_DB_FNAME_LEN] = {0};
|
||||
tNameGetFullDbName(pTableName, db);
|
||||
(void)tNameGetFullDbName(pTableName, db);
|
||||
|
||||
if (IS_SYS_DBNAME(pTableName->dbname)) {
|
||||
pVgroup->vgId = MNODE_HANDLE;
|
||||
if (pMgmtEps) {
|
||||
memcpy(&pVgroup->epSet, pMgmtEps, sizeof(pVgroup->epSet));
|
||||
TAOS_MEMCPY(&pVgroup->epSet, pMgmtEps, sizeof(pVgroup->epSet));
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
|
@ -1191,7 +1199,7 @@ int32_t ctgGetVgInfoFromHashValue(SCatalog* pCtg, SEpSet* pMgmtEps, SDBVgInfo* d
|
|||
|
||||
SVgroupInfo* vgInfo = NULL;
|
||||
char tbFullName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(pTableName, tbFullName);
|
||||
(void)tNameExtractFullName(pTableName, tbFullName);
|
||||
|
||||
uint32_t hashValue = taosGetTbHashVal(tbFullName, (uint32_t)strlen(tbFullName), dbInfo->hashMethod,
|
||||
dbInfo->hashPrefix, dbInfo->hashSuffix);
|
||||
|
|
@ -1243,28 +1251,41 @@ int32_t ctgGetVgInfosFromHashValue(SCatalog* pCtg, SEpSet* pMgmgEpSet, SCtgTaskR
|
|||
SVgroupInfo mgmtInfo = {0};
|
||||
mgmtInfo.vgId = MNODE_HANDLE;
|
||||
if (pMgmgEpSet) {
|
||||
memcpy(&mgmtInfo.epSet, pMgmgEpSet, sizeof(mgmtInfo.epSet));
|
||||
TAOS_MEMCPY(&mgmtInfo.epSet, pMgmgEpSet, sizeof(mgmtInfo.epSet));
|
||||
}
|
||||
|
||||
for (int32_t i = 0; i < tbNum; ++i) {
|
||||
vgInfo = taosMemoryMalloc(sizeof(SVgroupInfo));
|
||||
if (NULL == vgInfo) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
memcpy(vgInfo, &mgmtInfo, sizeof(mgmtInfo));
|
||||
TAOS_MEMCPY(vgInfo, &mgmtInfo, sizeof(mgmtInfo));
|
||||
|
||||
ctgDebug("Got tb hash vgroup, vgId:%d, epNum %d, current %s port %d", vgInfo->vgId, vgInfo->epSet.numOfEps,
|
||||
vgInfo->epSet.eps[vgInfo->epSet.inUse].fqdn, vgInfo->epSet.eps[vgInfo->epSet.inUse].port);
|
||||
|
||||
if (update) {
|
||||
SCtgFetch* pFetch = taosArrayGet(pCtx->pFetchs, tReq->msgIdx);
|
||||
if (NULL == pFetch) {
|
||||
ctgError("fail to get the %dth SCtgFetch, total:%d", tReq->msgIdx, (int32_t)taosArrayGetSize(pCtx->pFetchs));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
SMetaRes* pRes = taosArrayGet(pCtx->pResList, pFetch->resIdx + i);
|
||||
if (NULL == pFetch) {
|
||||
ctgError("fail to get the %dth SMetaRes, total:%d", pFetch->resIdx + i, (int32_t)taosArrayGetSize(pCtx->pResList));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
pRes->pRes = vgInfo;
|
||||
} else {
|
||||
res.pRes = vgInfo;
|
||||
taosArrayPush(pCtx->pResList, &res);
|
||||
if (NULL == taosArrayPush(pCtx->pResList, &res)) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
|
@ -1281,18 +1302,35 @@ int32_t ctgGetVgInfosFromHashValue(SCatalog* pCtg, SEpSet* pMgmgEpSet, SCtgTaskR
|
|||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
*vgInfo = *(SVgroupInfo*)taosArrayGet(dbInfo->vgArray, 0);
|
||||
SVgroupInfo* pSrcVg = (SVgroupInfo*)taosArrayGet(dbInfo->vgArray, 0);
|
||||
if (NULL == pSrcVg) {
|
||||
ctgError("fail to get the 0th SVgroupInfo, total:%d", (int32_t)taosArrayGetSize(dbInfo->vgArray));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
TAOS_MEMCPY(vgInfo, pSrcVg, sizeof(*pSrcVg));
|
||||
|
||||
ctgDebug("Got tb hash vgroup, vgId:%d, epNum %d, current %s port %d", vgInfo->vgId, vgInfo->epSet.numOfEps,
|
||||
vgInfo->epSet.eps[vgInfo->epSet.inUse].fqdn, vgInfo->epSet.eps[vgInfo->epSet.inUse].port);
|
||||
|
||||
if (update) {
|
||||
SCtgFetch* pFetch = taosArrayGet(pCtx->pFetchs, tReq->msgIdx);
|
||||
if (NULL == pFetch) {
|
||||
ctgError("fail to get the %dth SCtgFetch, total:%d", tReq->msgIdx, (int32_t)taosArrayGetSize(pCtx->pFetchs));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
SMetaRes* pRes = taosArrayGet(pCtx->pResList, pFetch->resIdx + i);
|
||||
if (NULL == pRes) {
|
||||
ctgError("fail to get the %dth SMetaRes, total:%d", pFetch->resIdx + i, (int32_t)taosArrayGetSize(pCtx->pResList));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
pRes->pRes = vgInfo;
|
||||
} else {
|
||||
res.pRes = vgInfo;
|
||||
taosArrayPush(pCtx->pResList, &res);
|
||||
if (NULL == taosArrayPush(pCtx->pResList, &res)) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1300,16 +1338,20 @@ int32_t ctgGetVgInfosFromHashValue(SCatalog* pCtg, SEpSet* pMgmgEpSet, SCtgTaskR
|
|||
}
|
||||
|
||||
char tbFullName[TSDB_TABLE_FNAME_LEN];
|
||||
sprintf(tbFullName, "%s.", dbFName);
|
||||
(void)sprintf(tbFullName, "%s.", dbFName);
|
||||
int32_t offset = strlen(tbFullName);
|
||||
SName* pName = NULL;
|
||||
int32_t tbNameLen = 0;
|
||||
|
||||
for (int32_t i = 0; i < tbNum; ++i) {
|
||||
pName = taosArrayGet(pNames, i);
|
||||
if (NULL == pName) {
|
||||
ctgError("fail to get the %dth SName, total:%d", i, (int32_t)taosArrayGetSize(pNames));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
tbNameLen = offset + strlen(pName->tname);
|
||||
strcpy(tbFullName + offset, pName->tname);
|
||||
TAOS_STRCPY(tbFullName + offset, pName->tname);
|
||||
|
||||
uint32_t hashValue = taosGetTbHashVal(tbFullName, (uint32_t)strlen(tbFullName), dbInfo->hashMethod,
|
||||
dbInfo->hashPrefix, dbInfo->hashSuffix);
|
||||
|
|
@ -1334,11 +1376,22 @@ int32_t ctgGetVgInfosFromHashValue(SCatalog* pCtg, SEpSet* pMgmgEpSet, SCtgTaskR
|
|||
|
||||
if (update) {
|
||||
SCtgFetch* pFetch = taosArrayGet(pCtx->pFetchs, tReq->msgIdx);
|
||||
if (NULL == pFetch) {
|
||||
ctgError("fail to get the %dth SCtgFetch, total:%d", tReq->msgIdx, (int32_t)taosArrayGetSize(pCtx->pFetchs));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
SMetaRes* pRes = taosArrayGet(pCtx->pResList, pFetch->resIdx + i);
|
||||
if (NULL == pRes) {
|
||||
ctgError("fail to get the %dth SMetaRes, total:%d", pFetch->resIdx + i, (int32_t)taosArrayGetSize(pCtx->pResList));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
pRes->pRes = pNewVg;
|
||||
} else {
|
||||
res.pRes = pNewVg;
|
||||
taosArrayPush(pCtx->pResList, &res);
|
||||
if (NULL == taosArrayPush(pCtx->pResList, &res)) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1351,7 +1404,6 @@ int32_t ctgGetVgIdsFromHashValue(SCatalog* pCtg, SDBVgInfo* dbInfo, char* dbFNam
|
|||
CTG_ERR_RET(ctgMakeVgArray(dbInfo));
|
||||
|
||||
int32_t vgNum = taosArrayGetSize(dbInfo->vgArray);
|
||||
|
||||
if (vgNum <= 0) {
|
||||
ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", dbFName, vgNum);
|
||||
CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
|
||||
|
|
@ -1359,11 +1411,11 @@ int32_t ctgGetVgIdsFromHashValue(SCatalog* pCtg, SDBVgInfo* dbInfo, char* dbFNam
|
|||
|
||||
SVgroupInfo* vgInfo = NULL;
|
||||
char tbFullName[TSDB_TABLE_FNAME_LEN];
|
||||
snprintf(tbFullName, sizeof(tbFullName), "%s.", dbFName);
|
||||
(void)snprintf(tbFullName, sizeof(tbFullName), "%s.", dbFName);
|
||||
int32_t offset = strlen(tbFullName);
|
||||
|
||||
for (int32_t i = 0; i < tbNum; ++i) {
|
||||
snprintf(tbFullName + offset, sizeof(tbFullName) - offset, "%s", pTbs[i]);
|
||||
(void)snprintf(tbFullName + offset, sizeof(tbFullName) - offset, "%s", pTbs[i]);
|
||||
uint32_t hashValue = taosGetTbHashVal(tbFullName, (uint32_t)strlen(tbFullName), dbInfo->hashMethod,
|
||||
dbInfo->hashPrefix, dbInfo->hashSuffix);
|
||||
|
||||
|
|
@ -1475,7 +1527,11 @@ int32_t ctgMakeVgArray(SDBVgInfo* dbInfo) {
|
|||
|
||||
void* pIter = taosHashIterate(dbInfo->vgHash, NULL);
|
||||
while (pIter) {
|
||||
taosArrayPush(dbInfo->vgArray, pIter);
|
||||
if (NULL == taosArrayPush(dbInfo->vgArray, pIter)) {
|
||||
taosHashCancelIterate(dbInfo->vgHash, pIter);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
pIter = taosHashIterate(dbInfo->vgHash, pIter);
|
||||
}
|
||||
|
||||
|
|
@ -1494,7 +1550,7 @@ int32_t ctgCloneVgInfo(SDBVgInfo* src, SDBVgInfo** dst) {
|
|||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
memcpy(*dst, src, sizeof(SDBVgInfo));
|
||||
TAOS_MEMCPY(*dst, src, sizeof(SDBVgInfo));
|
||||
|
||||
size_t hashSize = taosHashGetSize(src->vgHash);
|
||||
(*dst)->vgHash = taosHashInit(hashSize, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
|
||||
|
|
@ -1522,6 +1578,11 @@ int32_t ctgCloneVgInfo(SDBVgInfo* src, SDBVgInfo** dst) {
|
|||
|
||||
if (src->vgArray) {
|
||||
(*dst)->vgArray = taosArrayDup(src->vgArray, NULL);
|
||||
if (NULL == (*dst)->vgArray) {
|
||||
taosHashCleanup((*dst)->vgHash);
|
||||
taosMemoryFreeClear(*dst);
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
|
@ -1534,7 +1595,7 @@ int32_t ctgCloneMetaOutput(STableMetaOutput* output, STableMetaOutput** pOutput)
|
|||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
memcpy(*pOutput, output, sizeof(STableMetaOutput));
|
||||
TAOS_MEMCPY(*pOutput, output, sizeof(STableMetaOutput));
|
||||
|
||||
if (output->tbMeta) {
|
||||
int32_t metaSize = CTG_META_SIZE(output->tbMeta);
|
||||
|
|
@ -1542,6 +1603,7 @@ int32_t ctgCloneMetaOutput(STableMetaOutput* output, STableMetaOutput** pOutput)
|
|||
if (useCompress(output->tbMeta->tableType) && (*pOutput)->tbMeta->schemaExt) {
|
||||
schemaExtSize = output->tbMeta->tableInfo.numOfColumns * sizeof(SSchemaExt);
|
||||
}
|
||||
|
||||
(*pOutput)->tbMeta = taosMemoryMalloc(metaSize + schemaExtSize);
|
||||
qDebug("tbMeta cloned, size:%d, p:%p", metaSize, (*pOutput)->tbMeta);
|
||||
if (NULL == (*pOutput)->tbMeta) {
|
||||
|
|
@ -1550,10 +1612,10 @@ int32_t ctgCloneMetaOutput(STableMetaOutput* output, STableMetaOutput** pOutput)
|
|||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
memcpy((*pOutput)->tbMeta, output->tbMeta, metaSize);
|
||||
TAOS_MEMCPY((*pOutput)->tbMeta, output->tbMeta, metaSize);
|
||||
if (useCompress(output->tbMeta->tableType) && (*pOutput)->tbMeta->schemaExt) {
|
||||
(*pOutput)->tbMeta->schemaExt = (SSchemaExt *)((char *)(*pOutput)->tbMeta + metaSize);
|
||||
memcpy((*pOutput)->tbMeta->schemaExt, output->tbMeta->schemaExt, schemaExtSize);
|
||||
TAOS_MEMCPY((*pOutput)->tbMeta->schemaExt, output->tbMeta->schemaExt, schemaExtSize);
|
||||
} else {
|
||||
(*pOutput)->tbMeta->schemaExt = NULL;
|
||||
}
|
||||
|
|
@ -1576,8 +1638,18 @@ int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes) {
|
|||
|
||||
for (int32_t i = 0; i < num; ++i) {
|
||||
STableIndexInfo* pInfo = taosArrayGet(pIndex, i);
|
||||
if (NULL == pInfo) {
|
||||
qError("fail to get the %dth STableIndexInfo, total:%d", i, (int32_t)taosArrayGetSize(pIndex));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
pInfo = taosArrayPush(*pRes, pInfo);
|
||||
if (NULL == pInfo) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
pInfo->expr = taosStrdup(pInfo->expr);
|
||||
if (NULL == pInfo->expr) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
|
@ -1588,6 +1660,9 @@ int32_t ctgUpdateSendTargetInfo(SMsgSendInfo* pMsgSendInfo, int32_t msgType, cha
|
|||
pMsgSendInfo->target.type = TARGET_TYPE_VNODE;
|
||||
pMsgSendInfo->target.vgId = vgId;
|
||||
pMsgSendInfo->target.dbFName = taosStrdup(dbFName);
|
||||
if (NULL == pMsgSendInfo->target.dbFName) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
} else {
|
||||
pMsgSendInfo->target.type = TARGET_TYPE_MNODE;
|
||||
}
|
||||
|
|
@ -1604,6 +1679,11 @@ int32_t ctgGetTablesReqNum(SArray* pList) {
|
|||
int32_t n = taosArrayGetSize(pList);
|
||||
for (int32_t i = 0; i < n; ++i) {
|
||||
STablesReq* pReq = taosArrayGet(pList, i);
|
||||
if (NULL == pReq) {
|
||||
qError("fail to get the %dth STablesReq, total:%d", i, (int32_t)taosArrayGetSize(pList));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
total += taosArrayGetSize(pReq->pTables);
|
||||
}
|
||||
|
||||
|
|
@ -1613,6 +1693,9 @@ int32_t ctgGetTablesReqNum(SArray* pList) {
|
|||
int32_t ctgAddFetch(SArray** pFetchs, int32_t dbIdx, int32_t tbIdx, int32_t* fetchIdx, int32_t resIdx, int32_t flag) {
|
||||
if (NULL == (*pFetchs)) {
|
||||
*pFetchs = taosArrayInit(CTG_DEFAULT_FETCH_NUM, sizeof(SCtgFetch));
|
||||
if (NULL == *pFetchs) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
|
||||
SCtgFetch fetch = {0};
|
||||
|
|
@ -1622,7 +1705,9 @@ int32_t ctgAddFetch(SArray** pFetchs, int32_t dbIdx, int32_t tbIdx, int32_t* fet
|
|||
fetch.resIdx = resIdx;
|
||||
fetch.flag = flag;
|
||||
|
||||
taosArrayPush(*pFetchs, &fetch);
|
||||
if (NULL == taosArrayPush(*pFetchs, &fetch)) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
|
@ -1643,7 +1728,19 @@ int32_t ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch, SName** ppName) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static void* ctgCloneDbVgroup(void* pSrc) { return taosArrayDup((const SArray*)pSrc, NULL); }
|
||||
static int32_t ctgCloneDbVgroup(void* pSrc, void** ppDst) {
|
||||
#if 0
|
||||
if (NULL == pSrc) {
|
||||
*ppDst = NULL;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
*ppDst = taosArrayDup((const SArray*)pSrc, NULL);
|
||||
return (*ppDst) ? TSDB_CODE_SUCCESS : TSDB_CODE_OUT_OF_MEMORY;
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeDbVgroup(void* p) { taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); }
|
||||
|
||||
|
|
@ -1652,6 +1749,7 @@ int32_t ctgCloneDbCfgInfo(void* pSrc, SDbCfgInfo** ppDst) {
|
|||
if (NULL == pDst) {
|
||||
return terrno;
|
||||
}
|
||||
|
||||
TAOS_MEMCPY(pDst, pSrc, sizeof(SDbCfgInfo));
|
||||
if (((SDbCfgInfo *)pSrc)->pRetensions) {
|
||||
pDst->pRetensions = taosArrayDup(((SDbCfgInfo *)pSrc)->pRetensions, NULL);
|
||||
|
|
@ -1671,13 +1769,19 @@ static void ctgFreeDbCfgInfo(void* p) {
|
|||
freeDbCfgInfo(pDst);
|
||||
}
|
||||
|
||||
static void* ctgCloneDbInfo(void* pSrc) {
|
||||
static int32_t ctgCloneDbInfo(void* pSrc, void** ppDst) {
|
||||
#if 0
|
||||
SDbInfo* pDst = taosMemoryMalloc(sizeof(SDbInfo));
|
||||
if (NULL == pDst) {
|
||||
return NULL;
|
||||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
memcpy(pDst, pSrc, sizeof(SDbInfo));
|
||||
return pDst;
|
||||
|
||||
TAOS_MEMCPY(pDst, pSrc, sizeof(SDbInfo));
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeDbInfo(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
||||
|
|
@ -1710,55 +1814,83 @@ static void ctgFreeDbInfo(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
|||
|
||||
static void ctgFreeTableMeta(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneVgroupInfo(void* pSrc) {
|
||||
static int32_t ctgCloneVgroupInfo(void* pSrc, void** ppDst) {
|
||||
#if 0
|
||||
SVgroupInfo* pDst = taosMemoryMalloc(sizeof(SVgroupInfo));
|
||||
if (NULL == pDst) {
|
||||
return NULL;
|
||||
}
|
||||
memcpy(pDst, pSrc, sizeof(SVgroupInfo));
|
||||
return pDst;
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeVgroupInfo(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneTableIndexs(void* pSrc) { return taosArrayDup((const SArray*)pSrc, NULL); }
|
||||
static int32_t ctgCloneTableIndexs(void* pSrc, void** ppDst) {
|
||||
#if 0
|
||||
return taosArrayDup((const SArray*)pSrc, NULL);
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeTableIndexs(void* p) { taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneFuncInfo(void* pSrc) {
|
||||
static int32_t ctgCloneFuncInfo(void* pSrc, void** ppDst) {
|
||||
#if 0
|
||||
SFuncInfo* pDst = taosMemoryMalloc(sizeof(SFuncInfo));
|
||||
if (NULL == pDst) {
|
||||
return NULL;
|
||||
}
|
||||
memcpy(pDst, pSrc, sizeof(SFuncInfo));
|
||||
return pDst;
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeFuncInfo(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneIndexInfo(void* pSrc) {
|
||||
static int32_t ctgCloneIndexInfo(void* pSrc) {
|
||||
#if 0
|
||||
SIndexInfo* pDst = taosMemoryMalloc(sizeof(SIndexInfo));
|
||||
if (NULL == pDst) {
|
||||
return NULL;
|
||||
}
|
||||
memcpy(pDst, pSrc, sizeof(SIndexInfo));
|
||||
return pDst;
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeIndexInfo(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneUserAuth(void* pSrc) {
|
||||
static int32_t ctgCloneUserAuth(void* pSrc) {
|
||||
#if 0
|
||||
bool* pDst = taosMemoryMalloc(sizeof(bool));
|
||||
if (NULL == pDst) {
|
||||
return NULL;
|
||||
}
|
||||
*pDst = *(bool*)pSrc;
|
||||
return pDst;
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeUserAuth(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneQnodeList(void* pSrc) { return taosArrayDup((const SArray*)pSrc, NULL); }
|
||||
static int32_t ctgCloneQnodeList(void* pSrc) {
|
||||
#if 0
|
||||
return taosArrayDup((const SArray*)pSrc, NULL);
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeQnodeList(void* p) { taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); }
|
||||
|
||||
|
|
@ -1773,11 +1905,18 @@ static void ctgFreeQnodeList(void* p) { taosArrayDestroy((SArray*)((SMetaRes*)p)
|
|||
|
||||
static void ctgFreeTableCfg(void* p) { taosMemoryFree(((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneDnodeList(void* pSrc) { return taosArrayDup((const SArray*)pSrc, NULL); }
|
||||
static int32_t ctgCloneDnodeList(void* pSrc) {
|
||||
#if 0
|
||||
return taosArrayDup((const SArray*)pSrc, NULL);
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeDnodeList(void* p) { taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); }
|
||||
|
||||
static void* ctgCloneViewMeta(void* pSrc) {
|
||||
static int32_t ctgCloneViewMeta(void* pSrc) {
|
||||
#if 0
|
||||
SViewMeta* pSrcMeta = pSrc;
|
||||
SViewMeta* pDst = taosMemoryMalloc(sizeof(SViewMeta));
|
||||
if (NULL == pDst) {
|
||||
|
|
@ -1791,6 +1930,9 @@ static void* ctgCloneViewMeta(void* pSrc) {
|
|||
}
|
||||
memcpy(pDst->pSchema, pSrcMeta->pSchema, pSrcMeta->numOfCols * sizeof(*pSrcMeta->pSchema));
|
||||
return pDst;
|
||||
#else
|
||||
return TSDB_CODE_CTG_INTERNAL_ERROR;
|
||||
#endif
|
||||
}
|
||||
|
||||
static void ctgFreeViewMeta(void* p) {
|
||||
|
|
@ -1818,8 +1960,8 @@ int32_t ctgChkSetTbAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res) {
|
|||
|
||||
char tbFName[TSDB_TABLE_FNAME_LEN];
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameExtractFullName(&req->pRawReq->tbName, tbFName);
|
||||
tNameGetFullDbName(&req->pRawReq->tbName, dbFName);
|
||||
(void)tNameExtractFullName(&req->pRawReq->tbName, tbFName);
|
||||
(void)tNameGetFullDbName(&req->pRawReq->tbName, dbFName);
|
||||
|
||||
while (true) {
|
||||
taosMemoryFreeClear(pMeta);
|
||||
|
|
@ -1871,7 +2013,7 @@ int32_t ctgChkSetTbAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res) {
|
|||
continue;
|
||||
}
|
||||
|
||||
sprintf(tbFName, "%s.%s", dbFName, stbName);
|
||||
(void)sprintf(tbFName, "%s.%s", dbFName, stbName);
|
||||
continue;
|
||||
}
|
||||
|
||||
|
|
@ -1918,7 +2060,7 @@ int32_t ctgChkSetBasicAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res
|
|||
}
|
||||
|
||||
char dbFName[TSDB_DB_FNAME_LEN];
|
||||
tNameGetFullDbName(&pReq->tbName, dbFName);
|
||||
(void)tNameGetFullDbName(&pReq->tbName, dbFName);
|
||||
|
||||
// since that we add read/write previliges when create db, there is no need to check createdDbs
|
||||
#if 0
|
||||
|
|
@ -2002,9 +2144,9 @@ int32_t ctgChkSetViewAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res)
|
|||
|
||||
char viewFName[TSDB_VIEW_FNAME_LEN];
|
||||
if (IS_SYS_DBNAME(req->pRawReq->tbName.dbname)) {
|
||||
snprintf(viewFName, sizeof(viewFName), "%s.%s", req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname);
|
||||
(void)snprintf(viewFName, sizeof(viewFName), "%s.%s", req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname);
|
||||
} else {
|
||||
tNameExtractFullName(&req->pRawReq->tbName, viewFName);
|
||||
(void)tNameExtractFullName(&req->pRawReq->tbName, viewFName);
|
||||
}
|
||||
int32_t len = strlen(viewFName) + 1;
|
||||
|
||||
|
|
@ -2378,7 +2520,7 @@ void ctgGetGlobalCacheStat(SCtgCacheStat* pStat) {
|
|||
pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
|
||||
}
|
||||
|
||||
memcpy(pStat, &gCtgMgmt.statInfo.cache, sizeof(gCtgMgmt.statInfo.cache));
|
||||
TAOS_MEMCPY(pStat, &gCtgMgmt.statInfo.cache, sizeof(gCtgMgmt.statInfo.cache));
|
||||
}
|
||||
|
||||
void ctgGetGlobalCacheSize(uint64_t *pSize) {
|
||||
|
|
@ -2405,12 +2547,19 @@ int32_t ctgBuildViewNullRes(SCtgTask* pTask, SCtgViewsCtx* pCtx) {
|
|||
int32_t dbNum = taosArrayGetSize(pCtx->pNames);
|
||||
for (int32_t i = 0; i < dbNum; ++i) {
|
||||
STablesReq* pReq = taosArrayGet(pCtx->pNames, i);
|
||||
if (NULL == pReq) {
|
||||
qError("fail to get the %dth STablesReq, total:%d", i, (int32_t)taosArrayGetSize(pCtx->pNames));
|
||||
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
|
||||
}
|
||||
|
||||
int32_t viewNum = taosArrayGetSize(pReq->pTables);
|
||||
|
||||
ctgDebug("start to check views in db %s, viewNum %d", pReq->dbFName, viewNum);
|
||||
|
||||
for (int32_t m = 0; m < viewNum; ++m) {
|
||||
taosArrayPush(pCtx->pResList, &(SMetaData){0});
|
||||
if (NULL == taosArrayPush(pCtx->pResList, &(SMetaData){0})) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2435,17 +2584,27 @@ int32_t dupViewMetaFromRsp(SViewMetaRsp* pRsp, SViewMeta* pViewMeta) {
|
|||
if (pViewMeta->pSchema == NULL) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
memcpy(pViewMeta->pSchema, pRsp->pSchema, pViewMeta->numOfCols * sizeof(SSchema));
|
||||
|
||||
TAOS_MEMCPY(pViewMeta->pSchema, pRsp->pSchema, pViewMeta->numOfCols * sizeof(SSchema));
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
uint64_t ctgGetTbTSMACacheSize(STableTSMAInfo* pTsmaInfo) {
|
||||
if (!pTsmaInfo) return 0;
|
||||
if (!pTsmaInfo) {
|
||||
return 0;
|
||||
}
|
||||
uint64_t size = sizeof(STableTSMAInfo);
|
||||
if (pTsmaInfo->pFuncs) size += sizeof(STableTSMAFuncInfo) * pTsmaInfo->pFuncs->size;
|
||||
if (pTsmaInfo->pTags) size += sizeof(SSchema) * pTsmaInfo->pTags->size;
|
||||
if (pTsmaInfo->pUsedCols) size += sizeof(SSchema) * pTsmaInfo->pUsedCols->size;
|
||||
if (pTsmaInfo->pFuncs) {
|
||||
size += sizeof(STableTSMAFuncInfo) * pTsmaInfo->pFuncs->size;
|
||||
}
|
||||
if (pTsmaInfo->pTags) {
|
||||
size += sizeof(SSchema) * pTsmaInfo->pTags->size;
|
||||
}
|
||||
if (pTsmaInfo->pUsedCols) {
|
||||
size += sizeof(SSchema) * pTsmaInfo->pUsedCols->size;
|
||||
}
|
||||
|
||||
return size;
|
||||
}
|
||||
|
||||
|
|
@ -2455,8 +2614,14 @@ bool hasOutOfDateTSMACache(SArray* pTsmas) {
|
|||
}
|
||||
for (int32_t i = 0; i < pTsmas->size; ++i) {
|
||||
STSMACache* pTsmaInfo = taosArrayGetP(pTsmas, i);
|
||||
if (isCtgTSMACacheOutOfDate(pTsmaInfo)) return true;
|
||||
if (NULL == pTsmaInfo) {
|
||||
ASSERT(0);
|
||||
}
|
||||
if (isCtgTSMACacheOutOfDate(pTsmaInfo)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -2477,6 +2642,9 @@ int32_t ctgAddTSMAFetch(SArray** pFetchs, int32_t dbIdx, int32_t tbIdx, int32_t*
|
|||
CTG_TSMA_FETCH_TYPE fetchType, const SName* sourceTbName) {
|
||||
if (NULL == (*pFetchs)) {
|
||||
*pFetchs = taosArrayInit(CTG_DEFAULT_FETCH_NUM, sizeof(SCtgTSMAFetch));
|
||||
if (NULL == *pFetchs) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
}
|
||||
|
||||
SCtgTSMAFetch fetch = {0};
|
||||
|
|
@ -2487,9 +2655,13 @@ int32_t ctgAddTSMAFetch(SArray** pFetchs, int32_t dbIdx, int32_t tbIdx, int32_t*
|
|||
|
||||
fetch.flag = flag;
|
||||
fetch.fetchType = fetchType;
|
||||
if (sourceTbName) fetch.tsmaSourceTbName = *sourceTbName;
|
||||
if (sourceTbName) {
|
||||
fetch.tsmaSourceTbName = *sourceTbName;
|
||||
}
|
||||
|
||||
taosArrayPush(*pFetchs, &fetch);
|
||||
if (NULL == taosArrayPush(*pFetchs, &fetch)) {
|
||||
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue