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

fix compile issue

This commit is contained in:
dapan1121 2022-06-13 15:36:28 +08:00
parent eff9f9be4d
commit 7a78d81232
12 changed files with 142 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/common/tmsg.h
View File

@ -1134,6 +1134,7 @@ void tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp);
int32_t tSerializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp); int32_t tSerializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp); int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
void tFreeSTableMetaRsp(STableMetaRsp* pRsp); void tFreeSTableMetaRsp(STableMetaRsp* pRsp);
void tFreeSTableIndexRsp(void *info);
typedef struct { typedef struct {
SArray* pMetaRsp; // Array of STableMetaRsp SArray* pMetaRsp; // Array of STableMetaRsp

2
include/libs/catalog/catalog.h
View File

@ -282,6 +282,8 @@ int32_t catalogGetIndexMeta(SCatalog* pCtg, SRequestConnInfo* pConn, const char*
int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, SArray** pRes); int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, SArray** pRes);
int32_t catalogUpdateTableIndex(SCatalog* pCtg, STableIndexRsp *pRsp);
int32_t catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* funcName, SFuncInfo* pInfo); int32_t catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* funcName, SFuncInfo* pInfo);
int32_t catalogChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass); int32_t catalogChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass);

1
source/common/src/tmsg.c
View File

@ -2758,7 +2758,6 @@ void tFreeSTableIndexRsp(void *info) {
taosArrayDestroyEx(pInfo->pIndex, tFreeSTableIndexInfo); taosArrayDestroyEx(pInfo->pIndex, tFreeSTableIndexInfo);
} }
void tFreeSSTbHbRsp(SSTbHbRsp *pRsp) { void tFreeSSTbHbRsp(SSTbHbRsp *pRsp) {
int32_t numOfMeta = taosArrayGetSize(pRsp->pMetaRsp); int32_t numOfMeta = taosArrayGetSize(pRsp->pMetaRsp);
for (int32_t i = 0; i < numOfMeta; ++i) { for (int32_t i = 0; i < numOfMeta; ++i) {

1
source/dnode/mnode/impl/inc/mndSma.h
View File

@ -26,6 +26,7 @@ int32_t mndInitSma(SMnode *pMnode);
void mndCleanupSma(SMnode *pMnode); void mndCleanupSma(SMnode *pMnode);
SSmaObj *mndAcquireSma(SMnode *pMnode, char *smaName); SSmaObj *mndAcquireSma(SMnode *pMnode, char *smaName);
void mndReleaseSma(SMnode *pMnode, SSmaObj *pSma); void mndReleaseSma(SMnode *pMnode, SSmaObj *pSma);
int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/dnode/mnode/impl/src/mndSma.c
View File

@ -872,7 +872,7 @@ static int32_t mndGetSma(SMnode *pMnode, SUserIndexReq *indexReq, SUserIndexRsp
return code; return code;
} }
static int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist) { int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist) {
int32_t code = 0; int32_t code = 0;
SSmaObj *pSma = NULL; SSmaObj *pSma = NULL;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;

9
source/dnode/mnode/impl/src/mndStb.c
View File

@ -27,6 +27,7 @@
#include "mndTrans.h" #include "mndTrans.h"
#include "mndUser.h" #include "mndUser.h"
#include "mndVgroup.h" #include "mndVgroup.h"
#include "mndSma.h"
#include "tname.h" #include "tname.h"
#define STB_VER_NUMBER 1 #define STB_VER_NUMBER 1
@ -1638,7 +1639,7 @@ static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {
} }
} else { } else {
mDebug("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName); mDebug("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
if (mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp) != 0) { if (mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp, NULL) != 0) {
goto _OVER; goto _OVER;
} }
} }
@ -1710,7 +1711,7 @@ int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t
tFreeSTableMetaRsp(&metaRsp); tFreeSTableMetaRsp(&metaRsp);
} }
if (pStbVersion->smaVer != smaVer) { if (pStbVersion->smaVer && pStbVersion->smaVer != smaVer) {
bool exist = false; bool exist = false;
char tbFName[TSDB_TABLE_FNAME_LEN]; char tbFName[TSDB_TABLE_FNAME_LEN];
STableIndexRsp indexRsp = {0}; STableIndexRsp indexRsp = {0};
@ -1722,8 +1723,8 @@ int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t
indexRsp.pIndex = NULL; indexRsp.pIndex = NULL;
} }
strcpy(indexRsp->dbFName, pStbVersion->dbFName); strcpy(indexRsp.dbFName, pStbVersion->dbFName);
strcpy(indexRsp->tbName, pStbVersion->stbName); strcpy(indexRsp.tbName, pStbVersion->stbName);
taosArrayPush(hbRsp.pIndexRsp, &indexRsp); taosArrayPush(hbRsp.pIndexRsp, &indexRsp);
} }

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

@ -491,8 +491,8 @@ void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache);
void ctgRUnlockVgInfo(SCtgDBCache *dbCache); void ctgRUnlockVgInfo(SCtgDBCache *dbCache);
int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist); int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist);
int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta); int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName); int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName);
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass); int32_t ctgChkAuthFromCache(SCatalog* pCtg, char* user, char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId); int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId);
int32_t ctgDropDbVgroupEnqueue(SCatalog* pCtg, const char *dbFName, bool syncReq); int32_t ctgDropDbVgroupEnqueue(SCatalog* pCtg, const char *dbFName, bool syncReq);
int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq); int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq);
@ -501,13 +501,18 @@ int32_t ctgUpdateVgroupEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId
int32_t ctgUpdateTbMetaEnqueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq); int32_t ctgUpdateTbMetaEnqueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq);
int32_t ctgUpdateUserEnqueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq); int32_t ctgUpdateUserEnqueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq);
int32_t ctgUpdateVgEpsetEnqueue(SCatalog* pCtg, char *dbFName, int32_t vgId, SEpSet* pEpSet); int32_t ctgUpdateVgEpsetEnqueue(SCatalog* pCtg, char *dbFName, int32_t vgId, SEpSet* pEpSet);
int32_t ctgUpdateTbIndexEnqueue(SCatalog* pCtg, STableIndex *pIndex, bool syncOp); int32_t ctgUpdateTbIndexEnqueue(SCatalog* pCtg, STableIndex **pIndex, bool syncOp);
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type); int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type);
int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size); int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size);
int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size); int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size);
int32_t ctgUpdateTbMetaToCache(SCatalog* pCtg, STableMetaOutput* pOut, bool syncReq); int32_t ctgUpdateTbMetaToCache(SCatalog* pCtg, STableMetaOutput* pOut, bool syncReq);
int32_t ctgStartUpdateThread(); int32_t ctgStartUpdateThread();
int32_t ctgRelaunchGetTbMetaTask(SCtgTask *pTask); int32_t ctgRelaunchGetTbMetaTask(SCtgTask *pTask);
void ctgReleaseVgInfoToCache(SCatalog* pCtg, SCtgDBCache *dbCache);
int32_t ctgReadTbIndexFromCache(SCatalog* pCtg, SName* pTableName, SArray** pRes);
int32_t ctgDropTbIndexEnqueue(SCatalog* pCtg, SName* pName, bool syncOp);
int32_t ctgOpDropTbIndex(SCtgCacheOperation *operation);
int32_t ctgOpUpdateTbIndex(SCtgCacheOperation *operation);
@ -516,7 +521,7 @@ int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SBuildU
int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SArray *out, SCtgTask* pTask); int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SArray *out, SCtgTask* pTask);
int32_t ctgGetDBCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask); int32_t ctgGetDBCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask);
int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *indexName, SIndexInfo *out, SCtgTask* pTask); int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *indexName, SIndexInfo *out, SCtgTask* pTask);
int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName* name, SArray** out, SCtgTask* pTask); int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *name, STableIndex* out, SCtgTask* pTask);
int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *funcName, SFuncInfo *out, SCtgTask* pTask); int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *funcName, SFuncInfo *out, SCtgTask* pTask);
int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask); int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask);
int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo *pConn, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask); int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo *pConn, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask);
@ -545,6 +550,7 @@ int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* targ
char *ctgTaskTypeStr(CTG_TASK_TYPE type); char *ctgTaskTypeStr(CTG_TASK_TYPE type);
int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask); int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask);
int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes); int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes);
void ctgFreeSTableIndex(void *info);
extern SCatalogMgmt gCtgMgmt; extern SCatalogMgmt gCtgMgmt;

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

@ -350,7 +350,7 @@ int32_t ctgChkAuth(SCatalog* pCtg, SRequestConnInfo *pConn, const char* user, co
*pass = false; *pass = false;
CTG_ERR_RET(ctgChkAuthFromCache(pCtg, user, dbFName, type, &inCache, pass)); CTG_ERR_RET(ctgChkAuthFromCache(pCtg, (char*)user, (char*)dbFName, type, &inCache, pass));
if (inCache) { if (inCache) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -382,15 +382,38 @@ _return:
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgGetTbIndex(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SArray** pRes) { int32_t ctgGetTbIndex(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, SArray** pRes) {
CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pTableName, pRes)); CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pTableName, pRes));
if (*pRes) { if (*pRes) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
CTG_ERR_RET(ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pRes, NULL)); STableIndex *pIndex = taosMemoryCalloc(1, sizeof(STableIndex));
if (NULL == pIndex) {
CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
int32_t code = 0;
CTG_ERR_JRET(ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pIndex, NULL));
SArray* pInfo = NULL;
CTG_ERR_JRET(ctgCloneTableIndex(pIndex->pIndex, &pInfo));
*pRes = pInfo;
CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pCtg, &pIndex, false));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
_return:
tFreeSTableIndexRsp(pIndex);
taosMemoryFree(pIndex);
taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
*pRes = NULL;
CTG_RET(code);
} }
@ -416,7 +439,7 @@ int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTabl
CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo)); CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo));
if (dbCache) { if (dbCache) {
vgHash = dbCache->vgInfo->vgHash; vgHash = dbCache->vgCache.vgInfo->vgHash;
} else { } else {
vgHash = vgInfo->vgHash; vgHash = vgInfo->vgHash;
} }
@ -640,9 +663,9 @@ int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* vers
CTG_API_LEAVE(TSDB_CODE_SUCCESS); CTG_API_LEAVE(TSDB_CODE_SUCCESS);
} }
*version = dbCache->vgInfo->vgVersion; *version = dbCache->vgCache.vgInfo->vgVersion;
*dbId = dbCache->dbId; *dbId = dbCache->dbId;
*tableNum = dbCache->vgInfo->numOfTable; *tableNum = dbCache->vgCache.vgInfo->numOfTable;
ctgReleaseVgInfoToCache(pCtg, dbCache); ctgReleaseVgInfoToCache(pCtg, dbCache);
@ -969,7 +992,7 @@ int32_t catalogGetTableHashVgroup(SCatalog *pCtg, SRequestConnInfo *pConn, const
SDBVgInfo *vgInfo = NULL; SDBVgInfo *vgInfo = NULL;
CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo)); CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo));
CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, vgInfo ? vgInfo : dbCache->vgInfo, pTableName, pVgroup)); CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, vgInfo ? vgInfo : dbCache->vgCache.vgInfo, pTableName, pVgroup));
_return: _return:
@ -1177,36 +1200,11 @@ int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo *pConn, const SNam
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT); CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
} }
CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pTableName, pRes));
if (*pRes) {
CTG_API_LEAVE(TSDB_CODE_SUCCESS);
}
STableIndex *pIndex = taosMemoryCalloc(1, sizeof(STableIndex));
if (NULL == pIndex) {
CTG_API_LEAVE(TSDB_CODE_OUT_OF_MEMORY);
}
int32_t code = 0; int32_t code = 0;
CTG_ERR_JRET(ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pIndex, NULL)); CTG_ERR_JRET(ctgGetTbIndex(pCtg, pConn, (SName*)pTableName, pRes));
SArray* pInfo = NULL;
CTG_ERR_JRET(ctgCloneTableIndex(pIndex->pIndex, &pInfo));
*pRes = pInfo;
CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pCtg, pTableName, &pIndex, false));
CTG_API_LEAVE(code);
_return: _return:
tFreeSTableIndexRsp(pIndex);
taosMemoryFree(pIndex);
taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
*pRes = NULL;
CTG_API_LEAVE(code); CTG_API_LEAVE(code);
} }

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

@ -685,7 +685,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache)); CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));
if (NULL != dbCache) { if (NULL != dbCache) {
SVgroupInfo vgInfo = {0}; SVgroupInfo vgInfo = {0};
CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, ctx->pName, &vgInfo)); CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, ctx->pName, &vgInfo));
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);
@ -1016,7 +1016,7 @@ int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) {
CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache)); CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));
if (dbCache) { if (dbCache) {
SVgroupInfo vgInfo = {0}; SVgroupInfo vgInfo = {0};
CTG_ERR_RET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, ctx->pName, &vgInfo)); CTG_ERR_RET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, ctx->pName, &vgInfo));
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);
@ -1064,7 +1064,7 @@ int32_t ctgLaunchGetDbVgTask(SCtgTask *pTask) {
CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, pCtx->dbFName, &dbCache)); CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, pCtx->dbFName, &dbCache));
if (NULL != dbCache) { if (NULL != dbCache) {
CTG_ERR_JRET(ctgGenerateVgList(pCtg, dbCache->vgInfo->vgHash, (SArray**)&pTask->res)); CTG_ERR_JRET(ctgGenerateVgList(pCtg, dbCache->vgCache.vgInfo->vgHash, (SArray**)&pTask->res));
CTG_ERR_JRET(ctgHandleTaskEnd(pTask, 0)); CTG_ERR_JRET(ctgHandleTaskEnd(pTask, 0));
} else { } else {
@ -1098,7 +1098,7 @@ int32_t ctgLaunchGetTbHashTask(SCtgTask *pTask) {
if (NULL == pTask->res) { if (NULL == pTask->res) {
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY); CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
} }
CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, pCtx->pName, (SVgroupInfo*)pTask->res)); CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, pCtx->pName, (SVgroupInfo*)pTask->res));
CTG_ERR_JRET(ctgHandleTaskEnd(pTask, 0)); CTG_ERR_JRET(ctgHandleTaskEnd(pTask, 0));
} else { } else {
@ -1171,9 +1171,9 @@ int32_t ctgLaunchGetDbInfoTask(SCtgTask *pTask) {
SDbInfo* pInfo = (SDbInfo*)pTask->res; SDbInfo* pInfo = (SDbInfo*)pTask->res;
CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, pCtx->dbFName, &dbCache)); CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, pCtx->dbFName, &dbCache));
if (NULL != dbCache) { if (NULL != dbCache) {
pInfo->vgVer = dbCache->vgInfo->vgVersion; pInfo->vgVer = dbCache->vgCache.vgInfo->vgVersion;
pInfo->dbId = dbCache->dbId; pInfo->dbId = dbCache->dbId;
pInfo->tbNum = dbCache->vgInfo->numOfTable; pInfo->tbNum = dbCache->vgCache.vgInfo->numOfTable;
} else { } else {
pInfo->vgVer = CTG_DEFAULT_INVALID_VERSION; pInfo->vgVer = CTG_DEFAULT_INVALID_VERSION;
} }

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

@ -64,8 +64,12 @@ SCtgOperation gCtgCacheOperation[CTG_OP_MAX] = {
CTG_OP_UPDATE_TB_INDEX, CTG_OP_UPDATE_TB_INDEX,
"update tbIndex", "update tbIndex",
ctgOpUpdateTbIndex ctgOpUpdateTbIndex
} },
{
CTG_OP_DROP_TB_INDEX,
"drop tbIndex",
ctgOpDropTbIndex
}
}; };
@ -305,6 +309,7 @@ _return:
int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist) { int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist) {
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
SCtgTbCache *tbCache = NULL;
ctgAcquireTbMetaFromCache(pCtg, dbFName, tbName, &dbCache, &tbCache); ctgAcquireTbMetaFromCache(pCtg, dbFName, tbName, &dbCache, &tbCache);
if (NULL == tbCache) { if (NULL == tbCache) {
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache); ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
@ -383,7 +388,7 @@ _return:
CTG_RET(code); CTG_RET(code);
} }
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid,
char *stbName) { char *stbName) {
*sver = -1; *sver = -1;
*tver = -1; *tver = -1;
@ -444,10 +449,10 @@ int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *
} }
int32_t ctgReadTbTypeFromCache(SCatalog* pCtg, const char* dbFName, const char *tableName, int32_t *tbType) { int32_t ctgReadTbTypeFromCache(SCatalog* pCtg, char* dbFName, char *tableName, int32_t *tbType) {
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
SCtgTbCache *tbCache = NULL; SCtgTbCache *tbCache = NULL;
ctgAcquireTbMetaFromCache(pCtg, dbFName, tableName, &dbCache, &tbCache); CTG_ERR_RET(ctgAcquireTbMetaFromCache(pCtg, dbFName, tableName, &dbCache, &tbCache));
if (NULL == tbCache) { if (NULL == tbCache) {
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache); ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -461,7 +466,7 @@ int32_t ctgReadTbTypeFromCache(SCatalog* pCtg, const char* dbFName, const char *
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgReadTbIndexFromCache(SCatalog* pCtg, const SName* pTableName, SArray** pRes) { int32_t ctgReadTbIndexFromCache(SCatalog* pCtg, SName* pTableName, SArray** pRes) {
int32_t code = 0; int32_t code = 0;
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
SCtgTbCache *tbCache = NULL; SCtgTbCache *tbCache = NULL;
@ -485,7 +490,7 @@ _return:
CTG_RET(code); CTG_RET(code);
} }
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass) { int32_t ctgChkAuthFromCache(SCatalog* pCtg, char* user, char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass) {
if (NULL == pCtg->userCache) { if (NULL == pCtg->userCache) {
ctgDebug("empty user auth cache, user:%s", user); ctgDebug("empty user auth cache, user:%s", user);
goto _return; goto _return;
@ -862,7 +867,7 @@ int32_t ctgUpdateTbIndexEnqueue(SCatalog* pCtg, STableIndex **pIndex, bool syncO
_return: _return:
taosArrayDestroyEx(*pIndex, tFreeSTableIndexInfo); taosArrayDestroyEx((*pIndex)->pIndex, tFreeSTableIndexInfo);
taosMemoryFreeClear(*pIndex); taosMemoryFreeClear(*pIndex);
taosMemoryFreeClear(msg); taosMemoryFreeClear(msg);
@ -893,7 +898,6 @@ int32_t ctgDropTbIndexEnqueue(SCatalog* pCtg, SName* pName, bool syncOp) {
_return: _return:
taosArrayDestroyEx(pIndex, tFreeSTableIndexInfo);
taosMemoryFreeClear(msg); taosMemoryFreeClear(msg);
CTG_RET(code); CTG_RET(code);
@ -1172,7 +1176,7 @@ int32_t ctgRemoveDBFromCache(SCatalog* pCtg, SCtgDBCache *dbCache, const char* d
CTG_LOCK(CTG_WRITE, &dbCache->dbLock); CTG_LOCK(CTG_WRITE, &dbCache->dbLock);
atomic_store_8(&dbCache->deleted, 1); atomic_store_8(&dbCache->deleted, 1);
ctgRemoveStbRent(pCtg, &dbCache); ctgRemoveStbRent(pCtg, dbCache);
ctgFreeDbCache(dbCache); ctgFreeDbCache(dbCache);
CTG_UNLOCK(CTG_WRITE, &dbCache->dbLock); CTG_UNLOCK(CTG_WRITE, &dbCache->dbLock);
@ -1251,6 +1255,8 @@ int32_t ctgUpdateRentStbVersion(SCatalog *pCtg, char* dbFName, char* tbName, uin
ctgDebug("db %s,%" PRIx64 " stb %s,%" PRIx64 " sver %d tver %d smaVer %d updated to stbRent", ctgDebug("db %s,%" PRIx64 " stb %s,%" PRIx64 " sver %d tver %d smaVer %d updated to stbRent",
dbFName, dbId, tbName, suid, metaRent.sversion, metaRent.tversion, metaRent.smaVer); dbFName, dbId, tbName, suid, metaRent.sversion, metaRent.tversion, metaRent.smaVer);
return TSDB_CODE_SUCCESS;
} }
@ -1330,32 +1336,40 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char* dbFName, char *tbName, STableIndex **index) { int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char* dbFName, char *tbName, STableIndex **index) {
if (NULL == dbCache->tbCache) { if (NULL == dbCache->tbCache) {
ctgFreeSTableIndex(*index);
taosMemoryFreeClear(*index); taosMemoryFreeClear(*index);
ctgError("db is dropping, dbId:%"PRIx64, dbCache->dbId); ctgError("db is dropping, dbId:%"PRIx64, dbCache->dbId);
CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED); CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
} }
STableIndex* pIndex = *index; STableIndex* pIndex = *index;
uint64_t suid = pIndex->suid;
SCtgTbCache* pCache = taosHashGet(dbCache->tbCache, tbName, strlen(tbName)); SCtgTbCache* pCache = taosHashGet(dbCache->tbCache, tbName, strlen(tbName));
if (NULL == pCache) { if (NULL == pCache) {
SCtgTbCache cache = {0}; SCtgTbCache cache = {0};
cache.pIndex = pIndex; cache.pIndex = pIndex;
if (taosHashPut(table->tbCache, tbName, strlen(tbName), &cache, sizeof(cache)) != 0) { if (taosHashPut(dbCache->tbCache, tbName, strlen(tbName), &cache, sizeof(cache)) != 0) {
ctgFreeSTableIndex(*index);
taosMemoryFreeClear(*index); taosMemoryFreeClear(*index);
ctgError("taosHashPut new tbCache failed, tbName:%s", tbName); ctgError("taosHashPut new tbCache failed, tbName:%s", tbName);
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR); CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
} }
*index = NULL; *index = NULL;
ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version, taosArrayGetSize(pIndex->pIndex)); ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version, (int32_t)taosArrayGetSize(pIndex->pIndex));
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, pIndex->suid, pCache));
if (suid) {
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, pIndex->suid, pCache));
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
if (pCache->pIndex) { if (pCache->pIndex) {
if (0 == suid) {
suid = pCache->pIndex->suid;
}
taosArrayDestroyEx(pCache->pIndex->pIndex, tFreeSTableIndexInfo); taosArrayDestroyEx(pCache->pIndex->pIndex, tFreeSTableIndexInfo);
taosMemoryFreeClear(pCache->pIndex); taosMemoryFreeClear(pCache->pIndex);
} }
@ -1363,9 +1377,11 @@ int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char* dbFNa
pCache->pIndex = pIndex; pCache->pIndex = pIndex;
*index = NULL; *index = NULL;
ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version, taosArrayGetSize(pIndex->pIndex)); ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version, (int32_t)taosArrayGetSize(pIndex->pIndex));
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, pIndex->suid, pCache)); if (suid) {
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, suid, pCache));
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1390,6 +1406,7 @@ int32_t ctgOpUpdateVgroup(SCtgCacheOperation *operation) {
SCtgUpdateVgMsg *msg = operation->data; SCtgUpdateVgMsg *msg = operation->data;
SDBVgInfo* dbInfo = msg->dbInfo; SDBVgInfo* dbInfo = msg->dbInfo;
char* dbFName = msg->dbFName; char* dbFName = msg->dbFName;
SCatalog* pCtg = msg->pCtg;
if (NULL == dbInfo->vgHash) { if (NULL == dbInfo->vgHash) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1611,7 +1628,7 @@ int32_t ctgOpDropTbMeta(SCtgCacheOperation *operation) {
} }
if (dbCache->dbId != msg->dbId) { if (dbCache->dbId != msg->dbId) {
ctgDebug("dbId %" PRIx64 " not match with curId %"PRIx64", dbFName:%s, tbName:%s"msg->dbId, dbCache->dbId, msg->dbFName, msg->tbName); ctgDebug("dbId %" PRIx64 " not match with curId %"PRIx64", dbFName:%s, tbName:%s", msg->dbId, dbCache->dbId, msg->dbFName, msg->tbName);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1746,9 +1763,39 @@ int32_t ctgOpUpdateTbIndex(SCtgCacheOperation *operation) {
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
CTG_ERR_JRET(ctgGetAddDBCache(pCtg, pIndex->dbFName, 0, &dbCache)); CTG_ERR_JRET(ctgGetAddDBCache(pCtg, pIndex->dbFName, 0, &dbCache));
if (NULL == dbCache) {
CTG_ERR_JRET(code); CTG_ERR_JRET(ctgWriteTbIndexToCache(pCtg, dbCache, pIndex->dbFName, pIndex->tbName, &pIndex));
_return:
if (pIndex) {
taosArrayDestroyEx(pIndex->pIndex, tFreeSTableIndexInfo);
taosMemoryFreeClear(pIndex);
} }
taosMemoryFreeClear(msg);
CTG_RET(code);
}
int32_t ctgOpDropTbIndex(SCtgCacheOperation *operation) {
int32_t code = 0;
SCtgDropTbIndexMsg *msg = operation->data;
SCatalog* pCtg = msg->pCtg;
SCtgDBCache *dbCache = NULL;
CTG_ERR_JRET(ctgGetDBCache(pCtg, msg->dbFName, &dbCache));
if (NULL == dbCache) {
return TSDB_CODE_SUCCESS;
}
STableIndex* pIndex = taosMemoryCalloc(1, sizeof(STableIndex));
if (NULL == pIndex) {
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
strcpy(pIndex->tbName, msg->tbName);
strcpy(pIndex->dbFName, msg->dbFName);
pIndex->version = -1;
CTG_ERR_JRET(ctgWriteTbIndexToCache(pCtg, dbCache, pIndex->dbFName, pIndex->tbName, &pIndex)); CTG_ERR_JRET(ctgWriteTbIndexToCache(pCtg, dbCache, pIndex->dbFName, pIndex->tbName, &pIndex));

18
source/libs/catalog/src/ctgDbg.c
View File

@ -266,11 +266,11 @@ int32_t ctgdGetStatNum(char *option, void *res) {
} }
int32_t ctgdGetTbMetaNum(SCtgDBCache *dbCache) { int32_t ctgdGetTbMetaNum(SCtgDBCache *dbCache) {
return dbCache->table.tbCache ? (int32_t)taosHashGetSize(dbCache->table.tbCache) : 0; return dbCache->tbCache ? (int32_t)taosHashGetSize(dbCache->tbCache) : 0;
} }
int32_t ctgdGetStbNum(SCtgDBCache *dbCache) { int32_t ctgdGetStbNum(SCtgDBCache *dbCache) {
return dbCache->table.stbCache ? (int32_t)taosHashGetSize(dbCache->table.stbCache) : 0; return dbCache->stbCache ? (int32_t)taosHashGetSize(dbCache->stbCache) : 0;
} }
int32_t ctgdGetRentNum(SCtgRentMgmt *rent) { int32_t ctgdGetRentNum(SCtgRentMgmt *rent) {
@ -363,17 +363,17 @@ void ctgdShowDBCache(SCatalog* pCtg, SHashObj *dbHash) {
dbFName = taosHashGetKey(pIter, &len); dbFName = taosHashGetKey(pIter, &len);
int32_t metaNum = dbCache->table.tbCache ? taosHashGetSize(dbCache->table.tbCache) : 0; int32_t metaNum = dbCache->tbCache ? taosHashGetSize(dbCache->tbCache) : 0;
int32_t stbNum = dbCache->table.stbCache ? taosHashGetSize(dbCache->table.stbCache) : 0; int32_t stbNum = dbCache->stbCache ? taosHashGetSize(dbCache->stbCache) : 0;
int32_t vgVersion = CTG_DEFAULT_INVALID_VERSION; int32_t vgVersion = CTG_DEFAULT_INVALID_VERSION;
int32_t hashMethod = -1; int32_t hashMethod = -1;
int32_t vgNum = 0; int32_t vgNum = 0;
if (dbCache->vgInfo) { if (dbCache->vgCache.vgInfo) {
vgVersion = dbCache->vgInfo->vgVersion; vgVersion = dbCache->vgCache.vgInfo->vgVersion;
hashMethod = dbCache->vgInfo->hashMethod; hashMethod = dbCache->vgCache.vgInfo->hashMethod;
if (dbCache->vgInfo->vgHash) { if (dbCache->vgCache.vgInfo->vgHash) {
vgNum = taosHashGetSize(dbCache->vgInfo->vgHash); vgNum = taosHashGetSize(dbCache->vgCache.vgInfo->vgHash);
} }
} }

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

@ -44,6 +44,16 @@ char *ctgTaskTypeStr(CTG_TASK_TYPE type) {
} }
} }
void ctgFreeSTableIndex(void *info) {
if (NULL == info) {
return;
}
STableIndex *pInfo = (STableIndex *)info;
taosArrayDestroyEx(pInfo->pIndex, tFreeSTableIndexInfo);
}
void ctgFreeSMetaData(SMetaData* pData) { void ctgFreeSMetaData(SMetaData* pData) {
taosArrayDestroy(pData->pTableMeta); taosArrayDestroy(pData->pTableMeta);
pData->pTableMeta = NULL; pData->pTableMeta = NULL;