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

Merge pull request #26664 from taosdata/fix/TD-31016.1 

enh: catalog return code validation
This commit is contained in:
dapan1121 2024-07-19 10:27:54 +08:00 committed by GitHub
parent 78e0762714 ad1d85bb13
commit bdfc168f80
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 2628 additions and 1159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/os/osDef.h
View File

@ -220,7 +220,7 @@ void syslog(int unused, const char *format, ...);
// Linux, length of name must <= 16 (the last '\0' included)
#define setThreadName(name) \
do { \
prctl(PR_SET_NAME, (name)); \
(void)prctl(PR_SET_NAME, (name)); \
} while (0)
#define getThreadName(name) \
do { \

1
include/os/osMemory.h
View File

@ -55,6 +55,7 @@ void *taosMemoryMallocAlign(uint32_t alignment, int64_t size);
#define TAOS_MEMSET(_s, _c, _n) ((void)memset(_s, _c, _n))
#define TAOS_MEMCPY(_d, _s, _n) ((void)memcpy(_d, _s, _n))
#define TAOS_MEMMOVE(_d, _s, _n) ((void)memmove(_d, _s, _n))
#define taosMemoryFreeClear(ptr) \
do { \

1
include/os/osString.h
View File

@ -60,6 +60,7 @@ typedef enum { M2C = 0, C2M } ConvType;
} while (0)
#define TAOS_STRCPY(_dst, _src) ((void)strcpy(_dst, _src))
#define TAOS_STRNCPY(_dst, _src, _size) ((void)strncpy(_dst, _src, _size))
char *tstrdup(const char *src);
int32_t taosUcs4len(TdUcs4 *ucs4);

16
source/libs/catalog/inc/catalogInt.h
View File

@ -25,6 +25,7 @@ extern "C" {
#include "tcommon.h"
#include "ttimer.h"
#include "tglobal.h"
#include "os.h"
#define CTG_DEFAULT_CACHE_CLUSTER_NUMBER 6
#define CTG_DEFAULT_CACHE_VGROUP_NUMBER 100
@ -669,12 +670,12 @@ typedef struct SCtgCacheItemInfo {
#define CTG_AUTH_READ(_t) ((_t) == AUTH_TYPE_READ || (_t) == AUTH_TYPE_READ_OR_WRITE)
#define CTG_AUTH_WRITE(_t) ((_t) == AUTH_TYPE_WRITE || (_t) == AUTH_TYPE_READ_OR_WRITE)
#define CTG_QUEUE_INC() atomic_add_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
#define CTG_QUEUE_DEC() atomic_sub_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
#define CTG_QUEUE_INC() (void)atomic_add_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
#define CTG_QUEUE_DEC() (void)atomic_sub_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
#define CTG_STAT_INC(_item, _n) atomic_add_fetch_64(&(_item), _n)
#define CTG_STAT_DEC(_item, _n) atomic_sub_fetch_64(&(_item), _n)
#define CTG_STAT_GET(_item) atomic_load_64(&(_item))
#define CTG_STAT_INC(_item, _n) (void)atomic_add_fetch_64(&(_item), _n)
#define CTG_STAT_DEC(_item, _n) (void)atomic_sub_fetch_64(&(_item), _n)
#define CTG_STAT_GET(_item) (void)atomic_load_64(&(_item))
#define CTG_STAT_API_INC(item, n) (CTG_STAT_INC(gCtgMgmt.statInfo.api.item, n))
#define CTG_STAT_RT_INC(item, n) (CTG_STAT_INC(gCtgMgmt.statInfo.runtime.item, n))
@ -971,7 +972,7 @@ int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq
int32_t ctgGetTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetasCtx* ctx, int32_t dbIdx,
int32_t* fetchIdx, int32_t baseResIdx, SArray* pList);
void* ctgCloneDbCfgInfo(void* pSrc);
int32_t ctgCloneDbCfgInfo(void* pSrc, SDbCfgInfo** ppDst);
int32_t ctgOpUpdateVgroup(SCtgCacheOperation* action);
int32_t ctgOpUpdateDbCfg(SCtgCacheOperation *operation);
@ -1112,7 +1113,7 @@ int32_t ctgRemoveTbMeta(SCatalog* pCtg, SName* pTableName);
int32_t ctgRemoveCacheUser(SCatalog* pCtg, SCtgUserAuth* pUser, const char* user);
int32_t ctgGetTbHashVgroup(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, SVgroupInfo* pVgroup,
bool* exists);
SName* ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch);
int32_t ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch, SName** ppName);
int32_t ctgdGetOneHandle(SCatalog** pHandle);
int ctgVgInfoComp(const void* lp, const void* rp);
int32_t ctgMakeVgArray(SDBVgInfo* dbInfo);
@ -1165,6 +1166,7 @@ int32_t ctgGetStreamProgressFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn,
void* bInput);
int32_t ctgAddTSMAFetch(SArray** pFetchs, int32_t dbIdx, int32_t tbIdx, int32_t* fetchIdx, int32_t resIdx, int32_t flag,
CTG_TSMA_FETCH_TYPE fetchType, const SName* sourceTbName);
void ctgFreeTask(SCtgTask* pTask, bool freeRes);
extern SCatalogMgmt gCtgMgmt;
extern SCtgDebug gCTGDebug;

212
source/libs/catalog/src/catalog.c
View File

@ -86,7 +86,7 @@ int32_t ctgRefreshDBVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char*
if (code) {
if (CTG_DB_NOT_EXIST(code) && (NULL != dbCache)) {
ctgDebug("db no longer exist, dbFName:%s, dbId:0x%" PRIx64, input.db, input.dbId);
ctgDropDbCacheEnqueue(pCtg, input.db, input.dbId);
CTG_ERR_RET(ctgDropDbCacheEnqueue(pCtg, input.db, input.dbId));
}
CTG_ERR_RET(code);
@ -116,8 +116,7 @@ int32_t ctgRefreshTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx*
if (CTG_FLAG_IS_SYS_DB(ctx->flag)) {
ctgDebug("will refresh tbmeta, supposed in information_schema, tbName:%s", tNameGetTableName(ctx->pName));
CTG_ERR_JRET(
ctgGetTbMetaFromMnodeImpl(pCtg, pConn, (char*)ctx->pName->dbname, (char*)ctx->pName->tname, output, NULL));
CTG_ERR_JRET(ctgGetTbMetaFromMnodeImpl(pCtg, pConn, (char*)ctx->pName->dbname, (char*)ctx->pName->tname, output, NULL));
} else if (CTG_FLAG_IS_STB(ctx->flag)) {
ctgDebug("will refresh tbmeta, supposed to be stb, tbName:%s", tNameGetTableName(ctx->pName));
@ -128,8 +127,7 @@ int32_t ctgRefreshTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx*
CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, ctx->pName, &vgroupInfo, output, NULL));
}
} else {
ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(ctx->pName),
ctx->flag);
ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(ctx->pName), ctx->flag);
// if get from vnode failed or no table meta, will not try mnode
CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, ctx->pName, &vgroupInfo, output, NULL));
@ -166,7 +164,7 @@ int32_t ctgRefreshTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx*
if (CTG_IS_META_NULL(output->metaType)) {
ctgError("no tbmeta got, tbName:%s", tNameGetTableName(ctx->pName));
ctgRemoveTbMetaFromCache(pCtg, ctx->pName, false);
CTG_ERR_JRET(ctgRemoveTbMetaFromCache(pCtg, ctx->pName, false));
CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
}
@ -216,7 +214,7 @@ int32_t ctgGetTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx* ctx
}
if (CTG_IS_META_BOTH(output->metaType)) {
memcpy(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
TAOS_MEMCPY(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
*pTableMeta = output->tbMeta;
goto _return;
@ -233,7 +231,7 @@ int32_t ctgGetTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx* ctx
taosMemoryFreeClear(output->tbMeta);
SName stbName = *ctx->pName;
strcpy(stbName.tname, output->tbName);
TAOS_STRCPY(stbName.tname, output->tbName);
SCtgTbMetaCtx stbCtx = {0};
stbCtx.flag = ctx->flag;
stbCtx.pName = &stbName;
@ -244,7 +242,7 @@ int32_t ctgGetTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx* ctx
continue;
}
memcpy(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));
TAOS_MEMCPY(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));
break;
}
@ -254,15 +252,15 @@ _return:
if (CTG_TABLE_NOT_EXIST(code) && ctx->tbInfo.inCache) {
char dbFName[TSDB_DB_FNAME_LEN] = {0};
if (CTG_FLAG_IS_SYS_DB(ctx->flag)) {
strcpy(dbFName, ctx->pName->dbname);
TAOS_STRCPY(dbFName, ctx->pName->dbname);
} else {
tNameGetFullDbName(ctx->pName, dbFName);
(void)tNameGetFullDbName(ctx->pName, dbFName);
}
if (TSDB_SUPER_TABLE == ctx->tbInfo.tbType) {
ctgDropStbMetaEnqueue(pCtg, dbFName, ctx->tbInfo.dbId, ctx->pName->tname, ctx->tbInfo.suid, false);
(void)ctgDropStbMetaEnqueue(pCtg, dbFName, ctx->tbInfo.dbId, ctx->pName->tname, ctx->tbInfo.suid, false); // already in error
} else {
ctgDropTbMetaEnqueue(pCtg, dbFName, ctx->tbInfo.dbId, ctx->pName->tname, false);
(void)ctgDropTbMetaEnqueue(pCtg, dbFName, ctx->tbInfo.dbId, ctx->pName->tname, false); // already in error
}
}
@ -285,18 +283,18 @@ int32_t ctgUpdateTbMeta(SCatalog* pCtg, STableMetaRsp* rspMsg, bool syncOp) {
int32_t code = 0;
strcpy(output->dbFName, rspMsg->dbFName);
TAOS_STRCPY(output->dbFName, rspMsg->dbFName);
output->dbId = rspMsg->dbId;
if (TSDB_CHILD_TABLE == rspMsg->tableType && NULL == rspMsg->pSchemas) {
strcpy(output->ctbName, rspMsg->tbName);
TAOS_STRCPY(output->ctbName, rspMsg->tbName);
SET_META_TYPE_CTABLE(output->metaType);
CTG_ERR_JRET(queryCreateCTableMetaFromMsg(rspMsg, &output->ctbMeta));
} else {
strcpy(output->tbName, rspMsg->tbName);
TAOS_STRCPY(output->tbName, rspMsg->tbName);
SET_META_TYPE_TABLE(output->metaType);
@ -348,14 +346,14 @@ int32_t ctgChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, SUserAuthInfo *pReq,
_return:
ctgUpdateUserEnqueue(pCtg, &req.authInfo, false);
(void)ctgUpdateUserEnqueue(pCtg, &req.authInfo, false); // cache update not fatal error
CTG_RET(code);
}
int32_t ctgGetTbType(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTableName, int32_t* tbType) {
char dbFName[TSDB_DB_FNAME_LEN];
tNameGetFullDbName(pTableName, dbFName);
(void)tNameGetFullDbName(pTableName, dbFName);
CTG_ERR_RET(ctgReadTbTypeFromCache(pCtg, dbFName, pTableName->tname, tbType));
if (*tbType > 0) {
return TSDB_CODE_SUCCESS;
@ -454,7 +452,11 @@ int32_t ctgGetTbTag(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTableName,
tagVal.type = TSDB_DATA_TYPE_JSON;
tagVal.pData = pJson;
tagVal.nData = strlen(pJson);
taosArrayPush(pTagVals, &tagVal);
if (NULL == taosArrayPush(pTagVals, &tagVal)) {
taosMemoryFree(pJson);
taosArrayDestroy(pTagVals);
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
} else {
CTG_ERR_JRET(tTagToValArray((const STag*)pCfg->pTags, &pTagVals));
}
@ -484,7 +486,7 @@ int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTabl
CTG_ERR_JRET(ctgGetTbMeta(pCtg, pConn, &ctx, &tbMeta));
char db[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(pTableName, db);
(void)tNameGetFullDbName(pTableName, db);
SHashObj* vgHash = NULL;
CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo, NULL));
@ -555,7 +557,7 @@ int32_t ctgGetTbHashVgroup(SCatalog* pCtg, SRequestConnInfo* pConn, const SName*
SCtgDBCache* dbCache = NULL;
int32_t code = 0;
char db[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(pTableName, db);
(void)tNameGetFullDbName(pTableName, db);
SDBVgInfo* vgInfo = NULL;
CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo, exists));
@ -590,7 +592,7 @@ int32_t ctgGetTbsHashVgId(SCatalog* pCtg, SRequestConnInfo* pConn, int32_t acctI
SCtgDBCache* dbCache = NULL;
int32_t code = 0;
char dbFName[TSDB_DB_FNAME_LEN] = {0};
snprintf(dbFName, TSDB_DB_FNAME_LEN, "%d.%s", acctId, pDb);
(void)snprintf(dbFName, TSDB_DB_FNAME_LEN, "%d.%s", acctId, pDb);
SDBVgInfo* vgInfo = NULL;
CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, dbFName, &dbCache, &vgInfo, NULL));
@ -615,7 +617,7 @@ _return:
int32_t ctgGetCachedTbVgMeta(SCatalog* pCtg, const SName* pTableName, SVgroupInfo* pVgroup, STableMeta** pTableMeta) {
int32_t code = 0;
char db[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(pTableName, db);
(void)tNameGetFullDbName(pTableName, db);
SCtgDBCache *dbCache = NULL;
SCtgTbCache *tbCache = NULL;
@ -688,8 +690,8 @@ _return:
void ctgProcessTimerEvent(void *param, void *tmrId) {
CTG_API_NENTER();
ctgdShowCacheInfo();
ctgdShowStatInfo();
(void)ctgdShowCacheInfo();
(void)ctgdShowStatInfo();
int32_t cacheMaxSize = atomic_load_32(&tsMetaCacheMaxSize);
if (cacheMaxSize >= 0) {
@ -703,7 +705,7 @@ void ctgProcessTimerEvent(void *param, void *tmrId) {
int32_t code = ctgClearCacheEnqueue(NULL, true, false, false, false);
if (code) {
qError("clear cache enqueue failed, error:%s", tstrerror(code));
taosTmrReset(ctgProcessTimerEvent, CTG_DEFAULT_CACHE_MON_MSEC, NULL, gCtgMgmt.timer, &gCtgMgmt.cacheTimer);
(void)taosTmrReset(ctgProcessTimerEvent, CTG_DEFAULT_CACHE_MON_MSEC, NULL, gCtgMgmt.timer, &gCtgMgmt.cacheTimer);
}
goto _return;
@ -711,7 +713,7 @@ void ctgProcessTimerEvent(void *param, void *tmrId) {
}
qTrace("reset catalog timer");
taosTmrReset(ctgProcessTimerEvent, CTG_DEFAULT_CACHE_MON_MSEC, NULL, gCtgMgmt.timer, &gCtgMgmt.cacheTimer);
(void)taosTmrReset(ctgProcessTimerEvent, CTG_DEFAULT_CACHE_MON_MSEC, NULL, gCtgMgmt.timer, &gCtgMgmt.cacheTimer);
_return:
@ -723,10 +725,8 @@ int32_t ctgGetDBCfg(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName
if (pDbCfg->cfgVersion < 0) {
CTG_ERR_RET(ctgGetDBCfgFromMnode(pCtg, pConn, dbFName, pDbCfg, NULL));
SDbCfgInfo *pCfg = ctgCloneDbCfgInfo(pDbCfg);
if (NULL == pCfg) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SDbCfgInfo *pCfg = NULL;
CTG_ERR_RET(ctgCloneDbCfgInfo(pDbCfg, &pCfg));
CTG_ERR_RET(ctgUpdateDbCfgEnqueue(pCtg, dbFName, pDbCfg->dbId, pCfg, false));
}
@ -735,6 +735,59 @@ int32_t ctgGetDBCfg(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName
}
int32_t ctgGetTbTsmas(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTableName, SArray** ppRes) {
STableTSMAInfoRsp tsmasRsp = {0};
int32_t code = ctgGetTbTSMAFromMnode(pCtg, pConn, pTableName, &tsmasRsp, NULL, TDMT_MND_GET_TABLE_TSMA);
if (code == TSDB_CODE_MND_SMA_NOT_EXIST) {
code = 0;
goto _return;
}
CTG_ERR_JRET(code);
*ppRes = tsmasRsp.pTsmas;
tsmasRsp.pTsmas = NULL;
for (int32_t i = 0; i < (*ppRes)->size; ++i) {
CTG_ERR_JRET(ctgUpdateTbTSMAEnqueue(pCtg, taosArrayGet((*ppRes), i), 0, false));
}
return TSDB_CODE_SUCCESS;
_return:
if (tsmasRsp.pTsmas) {
tFreeTableTSMAInfoRsp(&tsmasRsp);
}
CTG_RET(code);
}
int32_t ctgGetTsma(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTsmaName, STableTSMAInfo** pTsma) {
STableTSMAInfoRsp tsmaRsp = {0};
int32_t code = ctgGetTbTSMAFromMnode(pCtg, pConn, pTsmaName, &tsmaRsp, NULL, TDMT_MND_GET_TSMA);
if (code == TSDB_CODE_MND_SMA_NOT_EXIST) {
code = 0;
goto _return;
}
CTG_ERR_JRET(code);
ASSERT(tsmaRsp.pTsmas && tsmaRsp.pTsmas->size == 1);
*pTsma = taosArrayGetP(tsmaRsp.pTsmas, 0);
taosArrayDestroy(tsmaRsp.pTsmas);
tsmaRsp.pTsmas = NULL;
_return:
if (tsmaRsp.pTsmas) {
tFreeTableTSMAInfoRsp(&tsmaRsp);
}
CTG_RET(code);
}
int32_t catalogInit(SCatalogCfg* cfg) {
qDebug("catalogInit start");
if (gCtgMgmt.pCluster) {
@ -742,10 +795,10 @@ int32_t catalogInit(SCatalogCfg* cfg) {
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
TAOS_MEMSET(&gCtgMgmt, 0, sizeof(gCtgMgmt));
if (cfg) {
memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
TAOS_MEMCPY(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
if (gCtgMgmt.cfg.maxDBCacheNum == 0) {
gCtgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
@ -1129,7 +1182,7 @@ int32_t catalogUpdateTableIndex(SCatalog* pCtg, STableIndexRsp* pRsp) {
CTG_API_LEAVE(TSDB_CODE_OUT_OF_MEMORY);
}
memcpy(pIndex, pRsp, sizeof(STableIndex));
TAOS_MEMCPY(pIndex, pRsp, sizeof(STableIndex));
CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pCtg, &pIndex, false));
@ -1247,17 +1300,26 @@ int32_t catalogChkTbMetaVersion(SCatalog* pCtg, SRequestConnInfo* pConn, SArray*
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
int32_t code = TSDB_CODE_SUCCESS;
SName name = {0};
int32_t sver = 0;
int32_t tver = 0;
int32_t tbNum = taosArrayGetSize(pTables);
for (int32_t i = 0; i < tbNum; ++i) {
STbSVersion* pTb = (STbSVersion*)taosArrayGet(pTables, i);
if (NULL == pTb) {
ctgError("fail to get the %dth table, tbNum:%d", i, tbNum);
CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
}
if (NULL == pTb->tbFName || 0 == pTb->tbFName[0]) {
continue;
}
tNameFromString(&name, pTb->tbFName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
if (tNameFromString(&name, pTb->tbFName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) {
ctgError("invalid tbFName format, tbFName:%s, idx:%d", pTb->tbFName, i);
CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
}
if (IS_SYS_DBNAME(name.dbname)) {
continue;
@ -1266,18 +1328,18 @@ int32_t catalogChkTbMetaVersion(SCatalog* pCtg, SRequestConnInfo* pConn, SArray*
int32_t tbType = 0;
uint64_t suid = 0;
char stbName[TSDB_TABLE_FNAME_LEN];
ctgReadTbVerFromCache(pCtg, &name, &sver, &tver, &tbType, &suid, stbName);
CTG_ERR_JRET(ctgReadTbVerFromCache(pCtg, &name, &sver, &tver, &tbType, &suid, stbName));
if ((sver >= 0 && sver < pTb->sver) || (tver >= 0 && tver < pTb->tver)) {
switch (tbType) {
case TSDB_CHILD_TABLE: {
SName stb = name;
tstrncpy(stb.tname, stbName, sizeof(stb.tname));
ctgRemoveTbMeta(pCtg, &stb);
CTG_ERR_JRET(ctgRemoveTbMeta(pCtg, &stb));
break;
}
case TSDB_SUPER_TABLE:
case TSDB_NORMAL_TABLE:
ctgRemoveTbMeta(pCtg, &name);
CTG_ERR_JRET(ctgRemoveTbMeta(pCtg, &name));
break;
default:
ctgError("ignore table type %d", tbType);
@ -1286,7 +1348,9 @@ int32_t catalogChkTbMetaVersion(SCatalog* pCtg, SRequestConnInfo* pConn, SArray*
}
}
CTG_API_LEAVE(TSDB_CODE_SUCCESS);
_return:
CTG_API_LEAVE(code);
}
int32_t catalogRefreshDBVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName) {
@ -1454,10 +1518,10 @@ int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo* pConn, const SC
_return:
if (pJob) {
taosReleaseRef(gCtgMgmt.jobPool, pJob->refId);
(void)taosReleaseRef(gCtgMgmt.jobPool, pJob->refId);
if (code) {
taosRemoveRef(gCtgMgmt.jobPool, pJob->refId);
(void)taosRemoveRef(gCtgMgmt.jobPool, pJob->refId);
}
}
@ -1558,7 +1622,7 @@ int32_t catalogGetExpiredUsers(SCatalog* pCtg, SUserAuthVersion** users, uint32_
while (pAuth != NULL) {
size_t len = 0;
void* key = taosHashGetKey(pAuth, &len);
strncpy((*users)[i].user, key, len);
TAOS_STRNCPY((*users)[i].user, key, len);
(*users)[i].user[len] = 0;
(*users)[i].version = pAuth->userAuth.version;
++i;
@ -1795,10 +1859,12 @@ int32_t catalogAsyncUpdateTSMA(SCatalog* pCtg, STableTSMAInfo** ppTsma, int32_t
if (!pCtg || !ppTsma) {
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
int32_t code = 0;
CTG_ERR_JRET(ctgUpdateTbTSMAEnqueue(pCtg, ppTsma, tsmaVersion, false));
_return:
CTG_API_LEAVE(code);
}
@ -1807,10 +1873,12 @@ int32_t catalogUpdateTSMA(SCatalog* pCtg, STableTSMAInfo** pTsma) {
if (!pCtg || !pTsma) {
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
int32_t code = 0;
CTG_ERR_JRET(ctgUpdateTbTSMAEnqueue(pCtg, pTsma, 0, true));
_return:
CTG_API_LEAVE(code);
}
@ -1823,36 +1891,14 @@ int32_t catalogRemoveTSMA(SCatalog* pCtg, const STableTSMAInfo* pTsma) {
}
if (!pCtg->dbCache) {
return TSDB_CODE_SUCCESS;
}
CTG_ERR_JRET(ctgDropTbTSMAEnqueue(pCtg, pTsma, true));
_return:
CTG_API_LEAVE(code);
}
int32_t ctgGetTbTsmas(SCatalog* pCtg, SRequestConnInfo* pConn, SName* pTableName, SArray** ppRes) {
STableTSMAInfoRsp tsmasRsp = {0};
int32_t code = ctgGetTbTSMAFromMnode(pCtg, pConn, pTableName, &tsmasRsp, NULL, TDMT_MND_GET_TABLE_TSMA);
if (code == TSDB_CODE_MND_SMA_NOT_EXIST) {
code = 0;
goto _return;
}
CTG_ERR_JRET(code);
assert(tsmasRsp.pTsmas);
assert(tsmasRsp.pTsmas->size > 0);
*ppRes = tsmasRsp.pTsmas;
tsmasRsp.pTsmas = NULL;
for (int32_t i = 0; i < (*ppRes)->size; ++i) {
CTG_ERR_JRET(ctgUpdateTbTSMAEnqueue(pCtg, taosArrayGet((*ppRes), i), 0, false));
}
return TSDB_CODE_SUCCESS;
CTG_ERR_JRET(ctgDropTbTSMAEnqueue(pCtg, pTsma, true));
_return:
if (tsmasRsp.pTsmas) {
tFreeTableTSMAInfoRsp(&tsmasRsp);
}
CTG_RET(code);
CTG_API_LEAVE(code);
}
int32_t catalogGetTableTsmas(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, SArray** pRes) {
@ -1870,27 +1916,6 @@ _return:
CTG_API_LEAVE(code);
}
int32_t ctgGetTsma(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTsmaName, STableTSMAInfo** pTsma) {
STableTSMAInfoRsp tsmaRsp = {0};
int32_t code = ctgGetTbTSMAFromMnode(pCtg, pConn, pTsmaName, &tsmaRsp, NULL, TDMT_MND_GET_TSMA);
if (code == TSDB_CODE_MND_SMA_NOT_EXIST) {
code = 0;
goto _return;
}
CTG_ERR_JRET(code);
ASSERT(tsmaRsp.pTsmas && tsmaRsp.pTsmas->size == 1);
*pTsma = taosArrayGetP(tsmaRsp.pTsmas, 0);
taosArrayDestroy(tsmaRsp.pTsmas);
tsmaRsp.pTsmas = NULL;
_return:
if (tsmaRsp.pTsmas) {
tFreeTableTSMAInfoRsp(&tsmaRsp);
}
CTG_RET(code);
}
int32_t catalogGetTsma(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTsmaName, STableTSMAInfo** pTsma) {
CTG_API_ENTER();
@ -1903,6 +1928,7 @@ int32_t catalogGetTsma(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTs
CTG_ERR_JRET(ctgGetTsma(pCtg, pConn, pTsmaName, pTsma));
_return:
CTG_API_LEAVE(code);
}
@ -1930,7 +1956,7 @@ void catalogDestroy(void) {
}
if (gCtgMgmt.cacheTimer) {
taosTmrStop(gCtgMgmt.cacheTimer);
(void)taosTmrStop(gCtgMgmt.cacheTimer);
gCtgMgmt.cacheTimer = NULL;
taosTmrCleanUp(gCtgMgmt.timer);
gCtgMgmt.timer = NULL;
@ -1939,8 +1965,8 @@ void catalogDestroy(void) {
atomic_store_8((int8_t*)&gCtgMgmt.exit, true);
if (!taosCheckCurrentInDll()) {
ctgClearCacheEnqueue(NULL, false, true, true, true);
taosThreadJoin(gCtgMgmt.updateThread, NULL);
(void)ctgClearCacheEnqueue(NULL, false, true, true, true);
(void)taosThreadJoin(gCtgMgmt.updateThread, NULL);
}
taosHashCleanup(gCtgMgmt.pCluster);

1656
source/libs/catalog/src/ctgAsync.c
View File

File diff suppressed because it is too large Load Diff

1239
source/libs/catalog/src/ctgCache.c
View File

File diff suppressed because it is too large Load Diff

241
source/libs/catalog/src/ctgRemote.c
View File

@ -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);
}
@ -556,12 +598,16 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
if (CTG_TASK_GET_TB_META_BATCH == pTask->type) {
SCtgTbMetasCtx* ctx = (SCtgTbMetasCtx*)pTask->taskCtx;
SCtgFetch* fetch = taosArrayGet(ctx->pFetchs, tReq->msgIdx);
pName = ctgGetFetchName(ctx->pNames, fetch);
CTG_ERR_JRET(ctgGetFetchName(ctx->pNames, fetch, &pName));
} else if (CTG_TASK_GET_TB_TSMA == pTask->type){
SCtgTbTSMACtx* pCtx = pTask->taskCtx;
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;
@ -616,12 +674,16 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
if (CTG_TASK_GET_TB_META_BATCH == pTask->type) {
SCtgTbMetasCtx* ctx = (SCtgTbMetasCtx*)pTask->taskCtx;
SCtgFetch* fetch = taosArrayGet(ctx->pFetchs, tReq->msgIdx);
pName = ctgGetFetchName(ctx->pNames, fetch);
CTG_ERR_JRET(ctgGetFetchName(ctx->pNames, fetch, &pName));
} else if (CTG_TASK_GET_TB_TSMA == pTask->type){
SCtgTbTSMACtx* pCtx = pTask->taskCtx;
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));

17
source/libs/catalog/src/ctgRent.c
View File

@ -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);

326
source/libs/catalog/src/ctgUtil.c
View File

@ -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,28 +1705,63 @@ 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;
}
SName* ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch) {
int32_t ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch, SName** ppName) {
STablesReq* pReq = (STablesReq*)taosArrayGet(pNames, pFetch->dbIdx);
return (SName*)taosArrayGet(pReq->pTables, pFetch->tbIdx);
if (NULL == pReq) {
qError("fail to get the %dth tb in pTables, tbNum:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pReq->pTables));
return TSDB_CODE_CTG_INTERNAL_ERROR;
}
*ppName = (SName*)taosArrayGet(pReq->pTables, pFetch->tbIdx);
if (NULL == *ppName) {
qError("fail to get the %dth tb in pTables, tbNum:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pReq->pTables));
return TSDB_CODE_CTG_INTERNAL_ERROR;
}
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); }
void* ctgCloneDbCfgInfo(void* pSrc) {
int32_t ctgCloneDbCfgInfo(void* pSrc, SDbCfgInfo** ppDst) {
SDbCfgInfo* pDst = taosMemoryMalloc(sizeof(SDbCfgInfo));
if (NULL == pDst) {
return NULL;
return terrno;
}
memcpy(pDst, pSrc, sizeof(SDbCfgInfo));
pDst->pRetensions = taosArrayDup(((SDbCfgInfo *)pSrc)->pRetensions, NULL);
return pDst;
TAOS_MEMCPY(pDst, pSrc, sizeof(SDbCfgInfo));
if (((SDbCfgInfo *)pSrc)->pRetensions) {
pDst->pRetensions = taosArrayDup(((SDbCfgInfo *)pSrc)->pRetensions, NULL);
if (NULL == pDst->pRetensions) {
taosMemoryFree(pDst);
return terrno;
}
}
*ppDst = pDst;
return TSDB_CODE_SUCCESS;
}
static void ctgFreeDbCfgInfo(void* p) {
@ -1651,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); }
@ -1690,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); }
@ -1753,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) {
@ -1771,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) {
@ -1798,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);
@ -1851,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;
}
@ -1898,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
@ -1982,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;
@ -2358,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) {
@ -2385,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);
}
}
}
@ -2415,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;
}
@ -2435,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;
}
@ -2457,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};
@ -2467,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;
}

76
source/libs/scheduler/test/schedulerTests.cpp
View File

@ -77,10 +77,10 @@ void schtInitLogFile() {
const int32_t maxLogFileNum = 10;
tsAsyncLog = 0;
qDebugFlag = 159;
strcpy(tsLogDir, TD_LOG_DIR_PATH);
TAOS_STRCPY(tsLogDir, TD_LOG_DIR_PATH);
if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
printf("failed to open log file in directory:%s\n", tsLogDir);
(void)printf("failed to open log file in directory:%s\n", tsLogDir);
}
}
@ -628,7 +628,7 @@ void schtFreeQueryJob(int32_t freeThread) {
schedulerFreeJob(&job, 0);
if (freeThread) {
if (++freeNum % schtTestPrintNum == 0) {
printf("FreeNum:%d\n", freeNum);
(void)printf("FreeNum:%d\n", freeNum);
}
}
}
@ -667,7 +667,7 @@ void *schtRunJobThread(void *aa) {
SQueryNodeLoad load = {0};
load.addr.epSet.numOfEps = 1;
strcpy(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
TAOS_STRCPY(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
load.addr.epSet.eps[0].port = 6031;
if (NULL == taosArrayPush(qnodeList, &load)) {
assert(0);
@ -800,7 +800,7 @@ void *schtRunJobThread(void *aa) {
schtFreeQueryDag(dag);
if (++jobFinished % schtTestPrintNum == 0) {
printf("jobFinished:%d\n", jobFinished);
(void)printf("jobFinished:%d\n", jobFinished);
}
++schtQueryId;
@ -834,9 +834,9 @@ TEST(queryTest, normalCase) {
SQueryNodeLoad load = {0};
load.addr.epSet.numOfEps = 1;
strcpy(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
TAOS_STRCPY(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
load.addr.epSet.eps[0].port = 6031;
taosArrayPush(qnodeList, &load);
assert(taosArrayPush(qnodeList, &load) != NULL);
int32_t code = schedulerInit();
ASSERT_EQ(code, 0);
@ -872,7 +872,7 @@ TEST(queryTest, normalCase) {
SDataBuf msg = {0};
void *rmsg = NULL;
schtBuildQueryRspMsg(&msg.len, &rmsg);
assert(0 == schtBuildQueryRspMsg(&msg.len, &rmsg));
msg.msgType = TDMT_SCH_QUERY_RSP;
msg.pData = rmsg;
@ -888,7 +888,7 @@ TEST(queryTest, normalCase) {
if (JOB_TASK_STATUS_EXEC == task->status) {
SDataBuf msg = {0};
void *rmsg = NULL;
schtBuildQueryRspMsg(&msg.len, &rmsg);
assert(0 == schtBuildQueryRspMsg(&msg.len, &rmsg));
msg.msgType = TDMT_SCH_QUERY_RSP;
msg.pData = rmsg;
@ -909,10 +909,10 @@ TEST(queryTest, normalCase) {
}
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
assert(0 == taosThreadAttrInit(&thattr));
TdThread thread1;
taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job);
assert(0 == taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job));
void *data = NULL;
req.syncReq = true;
@ -926,13 +926,13 @@ TEST(queryTest, normalCase) {
ASSERT_EQ(pRsp->numOfRows, 10);
taosMemoryFreeClear(data);
schReleaseJob(job);
(void)schReleaseJob(job);
schedulerDestroy();
schedulerFreeJob(&job, 0);
taosThreadJoin(thread1, NULL);
(void)taosThreadJoin(thread1, NULL);
}
TEST(queryTest, readyFirstCase) {
@ -948,9 +948,9 @@ TEST(queryTest, readyFirstCase) {
SQueryNodeLoad load = {0};
load.addr.epSet.numOfEps = 1;
strcpy(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
TAOS_STRCPY(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
load.addr.epSet.eps[0].port = 6031;
taosArrayPush(qnodeList, &load);
assert(NULL != taosArrayPush(qnodeList, &load));
int32_t code = schedulerInit();
ASSERT_EQ(code, 0);
@ -985,7 +985,7 @@ TEST(queryTest, readyFirstCase) {
SDataBuf msg = {0};
void *rmsg = NULL;
schtBuildQueryRspMsg(&msg.len, &rmsg);
assert(0 == schtBuildQueryRspMsg(&msg.len, &rmsg));
msg.msgType = TDMT_SCH_QUERY_RSP;
msg.pData = rmsg;
@ -1002,7 +1002,7 @@ TEST(queryTest, readyFirstCase) {
if (JOB_TASK_STATUS_EXEC == task->status) {
SDataBuf msg = {0};
void *rmsg = NULL;
schtBuildQueryRspMsg(&msg.len, &rmsg);
assert(0 == schtBuildQueryRspMsg(&msg.len, &rmsg));
msg.msgType = TDMT_SCH_QUERY_RSP;
msg.pData = rmsg;
@ -1023,10 +1023,10 @@ TEST(queryTest, readyFirstCase) {
}
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
assert(0 == taosThreadAttrInit(&thattr));
TdThread thread1;
taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job);
assert(0 == taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job));
void *data = NULL;
req.syncReq = true;
@ -1039,13 +1039,13 @@ TEST(queryTest, readyFirstCase) {
ASSERT_EQ(pRsp->numOfRows, 10);
taosMemoryFreeClear(data);
schReleaseJob(job);
(void)schReleaseJob(job);
schedulerDestroy();
schedulerFreeJob(&job, 0);
taosThreadJoin(thread1, NULL);
(void)taosThreadJoin(thread1, NULL);
}
TEST(queryTest, flowCtrlCase) {
@ -1065,9 +1065,9 @@ TEST(queryTest, flowCtrlCase) {
SQueryNodeLoad load = {0};
load.addr.epSet.numOfEps = 1;
strcpy(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
TAOS_STRCPY(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
load.addr.epSet.eps[0].port = 6031;
taosArrayPush(qnodeList, &load);
assert(NULL != taosArrayPush(qnodeList, &load));
int32_t code = schedulerInit();
ASSERT_EQ(code, 0);
@ -1078,7 +1078,7 @@ TEST(queryTest, flowCtrlCase) {
schtSetExecNode();
schtSetAsyncSendMsgToServer();
initTaskQueue();
assert(0 == initTaskQueue());
int32_t queryDone = 0;
SRequestConnInfo conn = {0};
@ -1106,7 +1106,7 @@ TEST(queryTest, flowCtrlCase) {
if (JOB_TASK_STATUS_EXEC == task->status && 0 != task->lastMsgType) {
SDataBuf msg = {0};
void *rmsg = NULL;
schtBuildQueryRspMsg(&msg.len, &rmsg);
assert(0 == schtBuildQueryRspMsg(&msg.len, &rmsg));
msg.msgType = TDMT_SCH_QUERY_RSP;
msg.pData = rmsg;
@ -1120,10 +1120,10 @@ TEST(queryTest, flowCtrlCase) {
}
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
assert(0 == taosThreadAttrInit(&thattr));
TdThread thread1;
taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job);
assert(0 == taosThreadCreate(&(thread1), &thattr, schtCreateFetchRspThread, &job));
void *data = NULL;
req.syncReq = true;
@ -1136,13 +1136,13 @@ TEST(queryTest, flowCtrlCase) {
ASSERT_EQ(pRsp->numOfRows, 10);
taosMemoryFreeClear(data);
schReleaseJob(job);
(void)schReleaseJob(job);
schedulerDestroy();
schedulerFreeJob(&job, 0);
taosThreadJoin(thread1, NULL);
(void)taosThreadJoin(thread1, NULL);
}
TEST(insertTest, normalCase) {
@ -1158,9 +1158,9 @@ TEST(insertTest, normalCase) {
SQueryNodeLoad load = {0};
load.addr.epSet.numOfEps = 1;
strcpy(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
TAOS_STRCPY(load.addr.epSet.eps[0].fqdn, "qnode0.ep");
load.addr.epSet.eps[0].port = 6031;
taosArrayPush(qnodeList, &load);
assert(NULL != taosArrayPush(qnodeList, &load));
int32_t code = schedulerInit();
ASSERT_EQ(code, 0);
@ -1171,12 +1171,12 @@ TEST(insertTest, normalCase) {
schtSetAsyncSendMsgToServer();
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
assert(0 == taosThreadAttrInit(&thattr));
schtJobDone = false;
TdThread thread1;
taosThreadCreate(&(thread1), &thattr, schtSendRsp, &insertJobRefId);
assert(0 == taosThreadCreate(&(thread1), &thattr, schtSendRsp, &insertJobRefId));
int32_t queryDone = 0;
SRequestConnInfo conn = {0};
@ -1204,17 +1204,17 @@ TEST(insertTest, normalCase) {
schedulerDestroy();
taosThreadJoin(thread1, NULL);
(void)taosThreadJoin(thread1, NULL);
}
TEST(multiThread, forceFree) {
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
assert(0 == taosThreadAttrInit(&thattr));
TdThread thread1, thread2, thread3;
taosThreadCreate(&(thread1), &thattr, schtRunJobThread, NULL);
assert(0 == taosThreadCreate(&(thread1), &thattr, schtRunJobThread, NULL));
// taosThreadCreate(&(thread2), &thattr, schtFreeJobThread, NULL);
taosThreadCreate(&(thread3), &thattr, schtFetchRspThread, NULL);
assert(0 == taosThreadCreate(&(thread3), &thattr, schtFetchRspThread, NULL));
while (true) {
if (schtTestDeadLoop) {
@ -1231,7 +1231,7 @@ TEST(multiThread, forceFree) {
TEST(otherTest, otherCase) {
// excpet test
schReleaseJob(0);
(void)schReleaseJob(0);
schFreeRpcCtx(NULL);
char* ep = NULL;