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

Merge pull request #10236 from taosdata/feature/qnode 

Feature/qnode
This commit is contained in:
dapan1121 2022-02-14 13:29:12 +08:00 committed by GitHub
parent 67f91c6d06 504d21553e
commit 32459c4ca2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 1607 additions and 852 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tmsg.h
View File

@ -2006,4 +2006,4 @@ static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* p
}
#endif
#endif /*_TD_COMMON_TAOS_MSG_H_*/
#endif /*_TD_COMMON_TAOS_MSG_H_*/

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

@ -30,7 +30,7 @@ extern "C" {
#include "tmsg.h"
#include "transport.h"
struct SCatalog;
typedef struct SCatalog SCatalog;
enum {
CTG_DBG_DB_NUM = 1,
@ -64,6 +64,7 @@ typedef struct SCatalogCfg {
typedef struct SSTableMetaVersion {
char dbFName[TSDB_DB_FNAME_LEN];
char stbName[TSDB_TABLE_NAME_LEN];
uint64_t dbId;
uint64_t suid;
int16_t sversion;
int16_t tversion;
@ -84,7 +85,7 @@ int32_t catalogInit(SCatalogCfg *cfg);
* @param catalogHandle (output, NO need to free it)
* @return error code
*/
int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle);
int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle);
/**
* Free a cluster's all catalog info, usually it's not necessary, until the application is closing.
@ -92,9 +93,9 @@ int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle);
* @param pCatalog (input, NO more usage)
* @return error code
*/
void catalogFreeHandle(struct SCatalog* pCatalog);
void catalogFreeHandle(SCatalog* pCatalog);
int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version);
int32_t catalogGetDBVgVersion(SCatalog* pCatalog, const char* dbName, int32_t* version);
/**
* Get a DB's all vgroup info.
@ -106,13 +107,13 @@ int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName,
* @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
* @return error code
*/
int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
int32_t catalogGetDBVgInfo(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, uint64_t dbId, SDBVgroupInfo* dbInfo);
int32_t catalogUpdateDBVgInfo(SCatalog* pCatalog, const char* dbName, uint64_t dbId, SDBVgInfo* dbInfo);
int32_t catalogRemoveDB(struct SCatalog* pCatalog, const char* dbName, uint64_t dbId);
int32_t catalogRemoveDB(SCatalog* pCatalog, const char* dbName, uint64_t dbId);
int32_t catalogRemoveSTableMeta(struct SCatalog* pCatalog, const char* dbName, const char* stbName, uint64_t suid);
int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid);
/**
* Get a table's meta data.
@ -123,7 +124,7 @@ int32_t catalogRemoveSTableMeta(struct SCatalog* pCatalog, const char* dbName, c
* @param pTableMeta(output, table meta data, NEED to free it by calller)
* @return error code
*/
int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
int32_t catalogGetTableMeta(SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
/**
* Get a super table's meta data.
@ -134,13 +135,13 @@ int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void * pTransporter, cons
* @param pTableMeta(output, table meta data, NEED to free it by calller)
* @return error code
*/
int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
int32_t catalogGetSTableMeta(SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg);
int32_t catalogUpdateSTableMeta(SCatalog* pCatalog, STableMetaRsp *rspMsg);
/**
* Force renew a table's local cached meta data.
* Force refresh a table's local cached meta data.
* @param pCatalog (input, got with catalogGetHandle)
* @param pTransporter (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
@ -148,10 +149,10 @@ int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg
* @param isSTable (input, is super table or not, 1:supposed to be stable, 0: supposed not to be stable, -1:not sure)
* @return error code
*/
int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable);
int32_t catalogRefreshTableMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable);
/**
* Force renew a table's local cached meta data and get the new one.
* Force refresh a table's local cached meta data and get the new one.
* @param pCatalog (input, got with catalogGetHandle)
* @param pTransporter (input, rpc object)
* @param pMgmtEps (input, mnode EPs)
@ -160,7 +161,7 @@ int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg
* @param isSTable (input, is super table or not, 1:supposed to be stable, 0: supposed not to be stable, -1:not sure)
* @return error code
*/
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable);
int32_t catalogRefreshGetTableMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable);
@ -173,7 +174,7 @@ int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg
* @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
* @return error code
*/
int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
int32_t catalogGetTableDistVgInfo(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
/**
* Get a table's vgroup from its name's hash value.
@ -184,7 +185,7 @@ int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pTransporter,
* @param vgInfo (output, vgroup info)
* @return error code
*/
int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pName, SVgroupInfo* vgInfo);
int32_t catalogGetTableHashVgroup(SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pName, SVgroupInfo* vgInfo);
/**
@ -196,14 +197,14 @@ int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void * pTransporter
* @param pRsp (output, response data)
* @return error code
*/
int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
int32_t catalogGetAllMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList);
int32_t catalogGetQnodeList(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList);
int32_t catalogGetExpiredSTables(struct SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
int32_t catalogGetExpiredDBs(struct SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num);
int32_t catalogGetExpiredDBs(SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num);
/**

6
include/libs/qcom/query.h
View File

@ -80,16 +80,16 @@ typedef struct STableMeta {
SSchema schema[];
} STableMeta;
typedef struct SDBVgroupInfo {
typedef struct SDBVgInfo {
int32_t vgVersion;
int8_t hashMethod;
SHashObj *vgHash; //key:vgId, value:SVgroupInfo
} SDBVgroupInfo;
} SDBVgInfo;
typedef struct SUseDbOutput {
char db[TSDB_DB_FNAME_LEN];
uint64_t dbId;
SDBVgroupInfo *dbVgroup;
SDBVgInfo *dbVgroup;
} SUseDbOutput;
enum {

10
source/client/src/clientHb.c
View File

@ -43,7 +43,7 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
if (rsp->vgVersion < 0) {
code = catalogRemoveDB(pCatalog, rsp->db, rsp->uid);
} else {
SDBVgroupInfo vgInfo = {0};
SDBVgInfo vgInfo = {0};
vgInfo.vgVersion = rsp->vgVersion;
vgInfo.hashMethod = rsp->hashMethod;
vgInfo.vgHash = taosHashInit(rsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
@ -68,10 +68,7 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
}
}
code = catalogUpdateDBVgroup(pCatalog, rsp->db, rsp->uid, &vgInfo);
if (code) {
taosHashCleanup(vgInfo.vgHash);
}
catalogUpdateDBVgInfo(pCatalog, rsp->db, rsp->uid, &vgInfo);
}
if (code) {
@ -94,13 +91,14 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
rsp->numOfColumns = ntohl(rsp->numOfColumns);
rsp->suid = be64toh(rsp->suid);
rsp->dbId = be64toh(rsp->dbId);
if (rsp->numOfColumns < 0) {
schemaNum = 0;
tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
catalogRemoveSTableMeta(pCatalog, rsp->dbFName, rsp->stbName, rsp->suid);
catalogRemoveStbMeta(pCatalog, rsp->dbFName, rsp->dbId, rsp->stbName, rsp->suid);
} else {
tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);

3
source/dnode/mgmt/impl/src/dndVnodes.c
View File

@ -382,7 +382,7 @@ static void *dnodeOpenVnodeFunc(void *param) {
pMgmt->openVnodes, pMgmt->totalVnodes);
dndReportStartup(pDnode, "open-vnodes", stepDesc);
SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId};
SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId, .dbId = pCfg->dbUid};
SVnode * pImpl = vnodeOpen(pCfg->path, &cfg);
if (pImpl == NULL) {
dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
@ -594,6 +594,7 @@ int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
vnodeCfg.pDnode = pDnode;
vnodeCfg.pTfs = pDnode->pTfs;
vnodeCfg.dbId = wrapperCfg.dbUid;
SVnode *pImpl = vnodeOpen(wrapperCfg.path, &vnodeCfg);
if (pImpl == NULL) {
dError("vgId:%d, failed to create vnode since %s", pCreate->vgId, terrstr());

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

@ -1523,4 +1523,4 @@ static int32_t mndRetrieveStb(SMnodeMsg *pReq, SShowObj *pShow, char *data, int3
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {
SSdb *pSdb = pMnode->pSdb;
sdbCancelFetch(pSdb, pIter);
}
}

1
source/dnode/vnode/inc/vnode.h
View File

@ -42,6 +42,7 @@ typedef struct STqCfg {
typedef struct SVnodeCfg {
int32_t vgId;
uint64_t dbId;
SDnode *pDnode;
STfs *pTfs;
uint64_t wsize;

1
source/dnode/vnode/src/vnd/vnodeMain.c
View File

@ -29,6 +29,7 @@ SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) {
cfg.vgId = pVnodeCfg->vgId;
cfg.pDnode = pVnodeCfg->pDnode;
cfg.pTfs = pVnodeCfg->pTfs;
cfg.dbId = pVnodeCfg->dbId;
}
// Validate options

3
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -93,6 +93,7 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
if (pTbCfg->type == META_CHILD_TABLE) {
pStbCfg = metaGetTbInfoByUid(pVnode->pMeta, pTbCfg->ctbCfg.suid);
if (pStbCfg == NULL) {
code = TSDB_CODE_VND_TB_NOT_EXIST;
goto _exit;
}
@ -116,9 +117,11 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
msgLen = sizeof(STableMetaRsp) + sizeof(SSchema) * (nCols + nTagCols);
pTbMetaMsg = (STableMetaRsp *)rpcMallocCont(msgLen);
if (pTbMetaMsg == NULL) {
code = TSDB_CODE_VND_OUT_OF_MEMORY;
goto _exit;
}
pTbMetaMsg->dbId = htobe64(pVnode->config.dbId);
memcpy(pTbMetaMsg->dbFName, pReq->dbFName, sizeof(pTbMetaMsg->dbFName));
strcpy(pTbMetaMsg->tbName, pReq->tbName);
if (pTbCfg->type == META_CHILD_TABLE) {

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

@ -27,7 +27,7 @@ extern "C" {
#define CTG_DEFAULT_CACHE_CLUSTER_NUMBER 6
#define CTG_DEFAULT_CACHE_VGROUP_NUMBER 100
#define CTG_DEFAULT_CACHE_DB_NUMBER 20
#define CTG_DEFAULT_CACHE_TABLEMETA_NUMBER 10000
#define CTG_DEFAULT_CACHE_TBLMETA_NUMBER 1000
#define CTG_DEFAULT_RENT_SECOND 10
#define CTG_DEFAULT_RENT_SLOT_SIZE 10
@ -47,9 +47,19 @@ enum {
CTG_RENT_STABLE,
};
enum {
CTG_ACT_UPDATE_VG = 0,
CTG_ACT_UPDATE_TBL,
CTG_ACT_REMOVE_DB,
CTG_ACT_REMOVE_STB,
CTG_ACT_REMOVE_TBL,
CTG_ACT_MAX
};
typedef struct SCtgDebug {
bool lockDebug;
bool cacheDebug;
bool apiDebug;
uint32_t showCachePeriodSec;
} SCtgDebug;
@ -65,7 +75,7 @@ typedef struct SCtgDBCache {
SRWLatch vgLock;
uint64_t dbId;
int8_t deleted;
SDBVgroupInfo *vgInfo;
SDBVgInfo *vgInfo;
SCtgTbMetaCache tbCache;
} SCtgDBCache;
@ -85,7 +95,6 @@ typedef struct SCtgRentMgmt {
typedef struct SCatalog {
uint64_t clusterId;
SRWLatch dbLock;
SHashObj *dbCache; //key:dbname, value:SCtgDBCache
SCtgRentMgmt dbRent;
SCtgRentMgmt stbRent;
@ -96,7 +105,8 @@ typedef struct SCtgApiStat {
} SCtgApiStat;
typedef struct SCtgRuntimeStat {
uint64_t qNum;
uint64_t qDoneNum;
} SCtgRuntimeStat;
typedef struct SCtgCacheStat {
@ -109,15 +119,70 @@ typedef struct SCatalogStat {
SCtgCacheStat cache;
} SCatalogStat;
typedef struct SCtgUpdateVgMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
uint64_t dbId;
SDBVgInfo* dbInfo;
} SCtgUpdateVgMsg;
typedef struct SCtgUpdateTblMsg {
SCatalog* pCtg;
STableMetaOutput* output;
} SCtgUpdateTblMsg;
typedef struct SCtgRemoveDBMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
uint64_t dbId;
} SCtgRemoveDBMsg;
typedef struct SCtgRemoveStbMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
char stbName[TSDB_TABLE_NAME_LEN];
uint64_t dbId;
uint64_t suid;
} SCtgRemoveStbMsg;
typedef struct SCtgMetaAction {
int32_t act;
void *data;
} SCtgMetaAction;
typedef struct SCtgQNode {
SCtgMetaAction action;
struct SCtgQNode *next;
} SCtgQNode;
typedef struct SCatalogMgmt {
bool exit;
SRWLatch lock;
SRWLatch qlock;
SCtgQNode *head;
SCtgQNode *tail;
tsem_t sem;
uint64_t qRemainNum;
pthread_t updateThread;
SHashObj *pCluster; //key: clusterId, value: SCatalog*
SCatalogStat stat;
SCatalogCfg cfg;
} SCatalogMgmt;
typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
typedef int32_t (*ctgActFunc)(SCtgMetaAction *);
typedef struct SCtgAction {
int32_t actId;
char name[32];
ctgActFunc func;
} SCtgAction;
#define CTG_QUEUE_ADD() atomic_add_fetch_64(&gCtgMgmt.qRemainNum, 1)
#define CTG_QUEUE_SUB() atomic_sub_fetch_64(&gCtgMgmt.qRemainNum, 1)
#define CTG_STAT_ADD(n) atomic_add_fetch_64(&(n), 1)
#define CTG_STAT_SUB(n) atomic_sub_fetch_64(&(n), 1)
#define CTG_IS_META_NULL(type) ((type) == META_TYPE_NULL_TABLE)
#define CTG_IS_META_CTABLE(type) ((type) == META_TYPE_CTABLE)
@ -130,20 +195,26 @@ typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
#define CTG_SET_STABLE(isSTable, tbType) do { (isSTable) = ((tbType) == TSDB_SUPER_TABLE) ? 1 : ((tbType) > TSDB_SUPER_TABLE ? 0 : -1); } while (0)
#define CTG_TBTYPE_MATCH(isSTable, tbType) (CTG_IS_UNKNOWN_STABLE(isSTable) || (CTG_IS_STABLE(isSTable) && (tbType) == TSDB_SUPER_TABLE) || (CTG_IS_NOT_STABLE(isSTable) && (tbType) != TSDB_SUPER_TABLE))
#define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST)
#define CTG_META_SIZE(pMeta) (sizeof(STableMeta) + ((pMeta)->tableInfo.numOfTags + (pMeta)->tableInfo.numOfColumns) * sizeof(SSchema))
#define ctgFatal(param, ...) qFatal("CTG:%p " param, pCatalog, __VA_ARGS__)
#define ctgError(param, ...) qError("CTG:%p " param, pCatalog, __VA_ARGS__)
#define ctgWarn(param, ...) qWarn("CTG:%p " param, pCatalog, __VA_ARGS__)
#define ctgInfo(param, ...) qInfo("CTG:%p " param, pCatalog, __VA_ARGS__)
#define ctgDebug(param, ...) qDebug("CTG:%p " param, pCatalog, __VA_ARGS__)
#define ctgTrace(param, ...) qTrace("CTG:%p " param, pCatalog, __VA_ARGS__)
#define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST)
#define CTG_DB_NOT_EXIST(code) (code == TSDB_CODE_MND_DB_NOT_EXIST)
#define ctgFatal(param, ...) qFatal("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgError(param, ...) qError("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgWarn(param, ...) qWarn("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgInfo(param, ...) qInfo("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgDebug(param, ...) qDebug("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgTrace(param, ...) qTrace("CTG:%p " param, pCtg, __VA_ARGS__)
#define CTG_LOCK_DEBUG(...) do { if (gCTGDebug.lockDebug) { qDebug(__VA_ARGS__); } } while (0)
#define CTG_CACHE_DEBUG(...) do { if (gCTGDebug.cacheDebug) { qDebug(__VA_ARGS__); } } while (0)
#define CTG_API_DEBUG(...) do { if (gCTGDebug.apiDebug) { qDebug(__VA_ARGS__); } } while (0)
#define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000
#define CTG_IS_LOCKED(_lock) atomic_load_32((_lock))
#define CTG_LOCK(type, _lock) do { \
if (CTG_READ == (type)) { \
assert(atomic_load_32((_lock)) >= 0); \
@ -181,8 +252,8 @@ typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
#define CTG_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define CTG_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#define CTG_API_ENTER() do { CTG_LOCK(CTG_READ, &ctgMgmt.lock); if (atomic_load_8(&ctgMgmt.exit)) { CTG_RET(TSDB_CODE_CTG_OUT_OF_SERVICE); } } while (0)
#define CTG_API_LEAVE(c) do { int32_t __code = c; CTG_UNLOCK(CTG_READ, &ctgMgmt.lock); CTG_RET(__code); } while (0)
#define CTG_API_LEAVE(c) do { int32_t __code = c; CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock); CTG_API_DEBUG("CTG API leave %s", __FUNCTION__); CTG_RET(__code); } while (0)
#define CTG_API_ENTER() do { CTG_API_DEBUG("CTG API enter %s", __FUNCTION__); CTG_LOCK(CTG_READ, &gCtgMgmt.lock); if (atomic_load_8(&gCtgMgmt.exit)) { CTG_API_LEAVE(TSDB_CODE_CTG_OUT_OF_SERVICE); } } while (0)

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

File diff suppressed because it is too large Load Diff

403
source/libs/catalog/test/catalogTests.cpp
View File

@ -33,18 +33,31 @@
#include "tep.h"
#include "trpc.h"
#include "tvariant.h"
#include "catalogInt.h"
namespace {
extern "C" int32_t ctgGetTableMetaFromCache(struct SCatalog *pCatalog, const SName *pTableName, STableMeta **pTableMeta,
int32_t *exist);
extern "C" int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *output);
extern "C" int32_t ctgDbgGetClusterCacheNum(struct SCatalog* pCatalog, int32_t type);
extern "C" int32_t ctgActUpdateTbl(SCtgMetaAction *action);
extern "C" int32_t ctgDbgEnableDebug(char *option);
extern "C" int32_t ctgDbgGetStatNum(char *option, void *res);
void ctgTestSetPrepareTableMeta();
void ctgTestSetPrepareCTableMeta();
void ctgTestSetPrepareSTableMeta();
void ctgTestSetPrepareMultiSTableMeta();
void ctgTestSetRspTableMeta();
void ctgTestSetRspCTableMeta();
void ctgTestSetRspSTableMeta();
void ctgTestSetRspMultiSTableMeta();
extern "C" SCatalogMgmt gCtgMgmt;
enum {
CTGT_RSP_VGINFO = 1,
CTGT_RSP_TBMETA,
CTGT_RSP_CTBMETA,
CTGT_RSP_STBMETA,
CTGT_RSP_MSTBMETA,
};
bool ctgTestStop = false;
bool ctgTestEnableSleep = false;
@ -68,6 +81,9 @@ char *ctgTestTablename = "table1";
char *ctgTestCTablename = "ctable1";
char *ctgTestSTablename = "stable1";
int32_t ctgTestRspFunc[10] = {0};
int32_t ctgTestRspIdx = 0;
void sendCreateDbMsg(void *shandle, SEpSet *pEpSet) {
SCreateDbReq *pReq = (SCreateDbReq *)rpcMallocCont(sizeof(SCreateDbReq));
strcpy(pReq->db, "1.db1");
@ -110,6 +126,8 @@ void ctgTestInitLogFile() {
tsAsyncLog = 0;
qDebugFlag = 159;
ctgDbgEnableDebug("api");
char temp[128] = {0};
sprintf(temp, "%s/%s", tsLogDir, defaultLogFileNamePrefix);
if (taosInitLog(temp, tsNumOfLogLines, maxLogFileNum) < 0) {
@ -176,11 +194,11 @@ void ctgTestBuildCTableMetaOutput(STableMetaOutput *output) {
strcpy(s->name, "tag1s");
}
void ctgTestBuildDBVgroup(SDBVgroupInfo **pdbVgroup) {
void ctgTestBuildDBVgroup(SDBVgInfo **pdbVgroup) {
static int32_t vgVersion = ctgTestVgVersion + 1;
int32_t vgNum = 0;
SVgroupInfo vgInfo = {0};
SDBVgroupInfo *dbVgroup = (SDBVgroupInfo *)calloc(1, sizeof(SDBVgroupInfo));
SDBVgInfo *dbVgroup = (SDBVgInfo *)calloc(1, sizeof(SDBVgInfo));
dbVgroup->vgVersion = vgVersion++;
@ -249,7 +267,7 @@ void ctgTestBuildSTableMetaRsp(STableMetaRsp *rspMsg) {
}
void ctgTestPrepareDbVgroups(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
void ctgTestRspDbVgroups(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
SUseDbRsp *rspMsg = NULL; // todo
pRsp->code = 0;
@ -285,7 +303,7 @@ void ctgTestPrepareDbVgroups(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcM
return;
}
void ctgTestPrepareTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
void ctgTestRspTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
pRsp->code = 0;
@ -321,7 +339,7 @@ void ctgTestPrepareTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcM
return;
}
void ctgTestPrepareCTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
void ctgTestRspCTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
pRsp->code = 0;
@ -364,7 +382,7 @@ void ctgTestPrepareCTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpc
return;
}
void ctgTestPrepareSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
void ctgTestRspSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
pRsp->code = 0;
@ -407,7 +425,7 @@ void ctgTestPrepareSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpc
return;
}
void ctgTestPrepareMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
void ctgTestRspMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
STableMetaRsp *rspMsg = NULL; // todo
static int32_t idx = 1;
@ -453,151 +471,193 @@ void ctgTestPrepareMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg,
return;
}
void ctgTestPrepareDbVgroupsAndNormalMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestPrepareDbVgroups(shandle, pEpSet, pMsg, pRsp);
ctgTestSetPrepareTableMeta();
void ctgTestRspByIdx(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
switch (ctgTestRspFunc[ctgTestRspIdx]) {
case CTGT_RSP_VGINFO:
ctgTestRspDbVgroups(shandle, pEpSet, pMsg, pRsp);
break;
case CTGT_RSP_TBMETA:
ctgTestRspTableMeta(shandle, pEpSet, pMsg, pRsp);
break;
case CTGT_RSP_CTBMETA:
ctgTestRspCTableMeta(shandle, pEpSet, pMsg, pRsp);
break;
case CTGT_RSP_STBMETA:
ctgTestRspSTableMeta(shandle, pEpSet, pMsg, pRsp);
break;
case CTGT_RSP_MSTBMETA:
ctgTestRspMultiSTableMeta(shandle, pEpSet, pMsg, pRsp);
break;
default:
break;
}
ctgTestRspIdx++;
return;
}
void ctgTestPrepareDbVgroupsAndChildMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestPrepareDbVgroups(shandle, pEpSet, pMsg, pRsp);
ctgTestSetPrepareCTableMeta();
void ctgTestRspDbVgroupsAndNormalMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestRspDbVgroups(shandle, pEpSet, pMsg, pRsp);
ctgTestSetRspTableMeta();
return;
}
void ctgTestPrepareDbVgroupsAndSuperMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestPrepareDbVgroups(shandle, pEpSet, pMsg, pRsp);
void ctgTestRspDbVgroupsAndChildMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestRspDbVgroups(shandle, pEpSet, pMsg, pRsp);
ctgTestSetPrepareSTableMeta();
ctgTestSetRspCTableMeta();
return;
}
void ctgTestPrepareDbVgroupsAndMultiSuperMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestPrepareDbVgroups(shandle, pEpSet, pMsg, pRsp);
void ctgTestRspDbVgroupsAndSuperMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestRspDbVgroups(shandle, pEpSet, pMsg, pRsp);
ctgTestSetPrepareMultiSTableMeta();
ctgTestSetRspSTableMeta();
return;
}
void ctgTestSetPrepareDbVgroups() {
void ctgTestRspDbVgroupsAndMultiSuperMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
ctgTestRspDbVgroups(shandle, pEpSet, pMsg, pRsp);
ctgTestSetRspMultiSTableMeta();
return;
}
void ctgTestSetRspDbVgroups() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareDbVgroups);
stub.set(rpcSendRecv, ctgTestRspDbVgroups);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareDbVgroups);
stub.set(f.second, ctgTestRspDbVgroups);
}
}
}
void ctgTestSetPrepareTableMeta() {
void ctgTestSetRspTableMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareTableMeta);
stub.set(rpcSendRecv, ctgTestRspTableMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareTableMeta);
stub.set(f.second, ctgTestRspTableMeta);
}
}
}
void ctgTestSetPrepareCTableMeta() {
void ctgTestSetRspCTableMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareCTableMeta);
stub.set(rpcSendRecv, ctgTestRspCTableMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareCTableMeta);
stub.set(f.second, ctgTestRspCTableMeta);
}
}
}
void ctgTestSetPrepareSTableMeta() {
void ctgTestSetRspSTableMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareSTableMeta);
stub.set(rpcSendRecv, ctgTestRspSTableMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareSTableMeta);
stub.set(f.second, ctgTestRspSTableMeta);
}
}
}
void ctgTestSetPrepareMultiSTableMeta() {
void ctgTestSetRspMultiSTableMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareMultiSTableMeta);
stub.set(rpcSendRecv, ctgTestRspMultiSTableMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareMultiSTableMeta);
stub.set(f.second, ctgTestRspMultiSTableMeta);
}
}
}
void ctgTestSetPrepareDbVgroupsAndNormalMeta() {
void ctgTestSetRspByIdx() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareDbVgroupsAndNormalMeta);
stub.set(rpcSendRecv, ctgTestRspByIdx);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareDbVgroupsAndNormalMeta);
stub.set(f.second, ctgTestRspByIdx);
}
}
}
void ctgTestSetPrepareDbVgroupsAndChildMeta() {
void ctgTestSetRspDbVgroupsAndNormalMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareDbVgroupsAndChildMeta);
stub.set(rpcSendRecv, ctgTestRspDbVgroupsAndNormalMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareDbVgroupsAndChildMeta);
stub.set(f.second, ctgTestRspDbVgroupsAndNormalMeta);
}
}
}
void ctgTestSetPrepareDbVgroupsAndSuperMeta() {
void ctgTestSetRspDbVgroupsAndChildMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareDbVgroupsAndSuperMeta);
stub.set(rpcSendRecv, ctgTestRspDbVgroupsAndChildMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareDbVgroupsAndSuperMeta);
stub.set(f.second, ctgTestRspDbVgroupsAndChildMeta);
}
}
}
void ctgTestSetPrepareDbVgroupsAndMultiSuperMeta() {
void ctgTestSetRspDbVgroupsAndSuperMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestPrepareDbVgroupsAndMultiSuperMeta);
stub.set(rpcSendRecv, ctgTestRspDbVgroupsAndSuperMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestPrepareDbVgroupsAndMultiSuperMeta);
stub.set(f.second, ctgTestRspDbVgroupsAndSuperMeta);
}
}
}
void ctgTestSetRspDbVgroupsAndMultiSuperMeta() {
static Stub stub;
stub.set(rpcSendRecv, ctgTestRspDbVgroupsAndMultiSuperMeta);
{
AddrAny any("libtransport.so");
std::map<std::string, void *> result;
any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
for (const auto &f : result) {
stub.set(f.second, ctgTestRspDbVgroupsAndMultiSuperMeta);
}
}
}
@ -612,7 +672,7 @@ void *ctgTestGetDbVgroupThread(void *param) {
int32_t n = 0;
while (!ctgTestStop) {
code = catalogGetDBVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
code = catalogGetDBVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
if (code) {
assert(0);
}
@ -635,12 +695,12 @@ void *ctgTestGetDbVgroupThread(void *param) {
void *ctgTestSetSameDbVgroupThread(void *param) {
struct SCatalog *pCtg = (struct SCatalog *)param;
int32_t code = 0;
SDBVgroupInfo *dbVgroup = NULL;
SDBVgInfo *dbVgroup = NULL;
int32_t n = 0;
while (!ctgTestStop) {
ctgTestBuildDBVgroup(&dbVgroup);
code = catalogUpdateDBVgroup(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
code = catalogUpdateDBVgInfo(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
if (code) {
assert(0);
}
@ -660,12 +720,12 @@ void *ctgTestSetSameDbVgroupThread(void *param) {
void *ctgTestSetDiffDbVgroupThread(void *param) {
struct SCatalog *pCtg = (struct SCatalog *)param;
int32_t code = 0;
SDBVgroupInfo *dbVgroup = NULL;
SDBVgInfo *dbVgroup = NULL;
int32_t n = 0;
while (!ctgTestStop) {
ctgTestBuildDBVgroup(&dbVgroup);
code = catalogUpdateDBVgroup(pCtg, ctgTestDbname, ctgTestDbId++, dbVgroup);
code = catalogUpdateDBVgInfo(pCtg, ctgTestDbname, ctgTestDbId++, dbVgroup);
if (code) {
assert(0);
}
@ -716,14 +776,24 @@ void *ctgTestGetCtableMetaThread(void *param) {
void *ctgTestSetCtableMetaThread(void *param) {
struct SCatalog *pCtg = (struct SCatalog *)param;
int32_t code = 0;
SDBVgroupInfo dbVgroup = {0};
SDBVgInfo dbVgroup = {0};
int32_t n = 0;
STableMetaOutput output = {0};
STableMetaOutput *output = NULL;
ctgTestBuildCTableMetaOutput(&output);
SCtgMetaAction action = {0};
action.act = CTG_ACT_UPDATE_TBL;
while (!ctgTestStop) {
code = ctgUpdateTableMetaCache(pCtg, &output);
output = (STableMetaOutput *)malloc(sizeof(STableMetaOutput));
ctgTestBuildCTableMetaOutput(output);
SCtgUpdateTblMsg *msg = (SCtgUpdateTblMsg *)malloc(sizeof(SCtgUpdateTblMsg));
msg->pCtg = pCtg;
msg->output = output;
action.data = msg;
code = ctgActUpdateTbl(&action);
if (code) {
assert(0);
}
@ -736,8 +806,6 @@ void *ctgTestSetCtableMetaThread(void *param) {
}
}
tfree(output.tbMeta);
return NULL;
}
@ -749,7 +817,7 @@ TEST(tableMeta, normalTable) {
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroups();
ctgTestSetRspDbVgroups();
initQueryModuleMsgHandle();
@ -770,7 +838,11 @@ TEST(tableMeta, normalTable) {
ASSERT_EQ(vgInfo.vgId, 8);
ASSERT_EQ(vgInfo.epset.numOfEps, 3);
ctgTestSetPrepareTableMeta();
while (0 == ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_DB_NUM)) {
usleep(10000);
}
ctgTestSetRspTableMeta();
STableMeta *tableMeta = NULL;
code = catalogGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta);
@ -784,6 +856,16 @@ TEST(tableMeta, normalTable) {
ASSERT_EQ(tableMeta->tableInfo.precision, 1);
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
while (true) {
uint32_t n = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM);
if (0 == n) {
usleep(10000);
} else {
break;
}
}
tableMeta = NULL;
code = catalogGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta);
ASSERT_EQ(code, 0);
@ -832,6 +914,7 @@ TEST(tableMeta, normalTable) {
ASSERT_EQ(allStbNum, 0);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(tableMeta, childTableCase) {
@ -841,7 +924,7 @@ TEST(tableMeta, childTableCase) {
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroupsAndChildMeta();
ctgTestSetRspDbVgroupsAndChildMeta();
initQueryModuleMsgHandle();
@ -868,6 +951,16 @@ TEST(tableMeta, childTableCase) {
ASSERT_EQ(tableMeta->tableInfo.precision, 1);
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
while (true) {
uint32_t n = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM);
if (0 == n) {
usleep(10000);
} else {
break;
}
}
tableMeta = NULL;
code = catalogGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta);
ASSERT_EQ(code, 0);
@ -930,6 +1023,7 @@ TEST(tableMeta, childTableCase) {
ASSERT_EQ(allStbNum, 1);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(tableMeta, superTableCase) {
@ -937,7 +1031,7 @@ TEST(tableMeta, superTableCase) {
void *mockPointer = (void *)0x1;
SVgroupInfo vgInfo = {0};
ctgTestSetPrepareDbVgroupsAndSuperMeta();
ctgTestSetRspDbVgroupsAndSuperMeta();
initQueryModuleMsgHandle();
@ -966,7 +1060,17 @@ TEST(tableMeta, superTableCase) {
ASSERT_EQ(tableMeta->tableInfo.precision, 1);
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
ctgTestSetPrepareCTableMeta();
while (true) {
uint32_t n = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM);
if (0 == n) {
usleep(10000);
} else {
break;
}
}
ctgTestSetRspCTableMeta();
tableMeta = NULL;
@ -983,8 +1087,18 @@ TEST(tableMeta, superTableCase) {
ASSERT_EQ(tableMeta->tableInfo.precision, 1);
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
while (true) {
uint32_t n = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM);
if (2 != n) {
usleep(10000);
} else {
break;
}
}
tableMeta = NULL;
code = catalogRenewAndGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta, 0);
code = catalogRefreshGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta, 0);
ASSERT_EQ(code, 0);
ASSERT_EQ(tableMeta->vgId, 9);
ASSERT_EQ(tableMeta->tableType, TSDB_CHILD_TABLE);
@ -1032,6 +1146,7 @@ TEST(tableMeta, superTableCase) {
ASSERT_EQ(allStbNum, 1);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(tableMeta, rmStbMeta) {
@ -1041,7 +1156,7 @@ TEST(tableMeta, rmStbMeta) {
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroupsAndSuperMeta();
ctgTestSetRspDbVgroupsAndSuperMeta();
initQueryModuleMsgHandle();
@ -1070,9 +1185,30 @@ TEST(tableMeta, rmStbMeta) {
ASSERT_EQ(tableMeta->tableInfo.precision, 1);
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
code = catalogRemoveSTableMeta(pCtg, "1.db1", ctgTestSTablename, ctgTestSuid);
while (true) {
uint32_t n = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM);
if (0 == n) {
usleep(10000);
} else {
break;
}
}
code = catalogRemoveStbMeta(pCtg, "1.db1", ctgTestDbId, ctgTestSTablename, ctgTestSuid);
ASSERT_EQ(code, 0);
while (true) {
int32_t n = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM);
int32_t m = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_STB_RENT_NUM);
if (n || m) {
usleep(10000);
} else {
break;
}
}
ASSERT_EQ(ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_DB_NUM), 1);
ASSERT_EQ(ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM), 0);
ASSERT_EQ(ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_STB_NUM), 0);
@ -1080,6 +1216,7 @@ TEST(tableMeta, rmStbMeta) {
ASSERT_EQ(ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_STB_RENT_NUM), 0);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(tableMeta, updateStbMeta) {
@ -1089,7 +1226,7 @@ TEST(tableMeta, updateStbMeta) {
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroupsAndSuperMeta();
ctgTestSetRspDbVgroupsAndSuperMeta();
initQueryModuleMsgHandle();
@ -1118,6 +1255,16 @@ TEST(tableMeta, updateStbMeta) {
ASSERT_EQ(tableMeta->tableInfo.precision, 1);
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
while (true) {
uint32_t n = ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM);
if (0 == n) {
usleep(10000);
} else {
break;
}
}
tfree(tableMeta);
STableMetaRsp rsp = {0};
@ -1126,6 +1273,16 @@ TEST(tableMeta, updateStbMeta) {
code = catalogUpdateSTableMeta(pCtg, &rsp);
ASSERT_EQ(code, 0);
while (true) {
uint64_t n = 0;
ctgDbgGetStatNum("runtime.qDoneNum", (void *)&n);
if (n != 3) {
usleep(10000);
} else {
break;
}
}
ASSERT_EQ(ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_DB_NUM), 1);
ASSERT_EQ(ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM), 1);
ASSERT_EQ(ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_STB_NUM), 1);
@ -1148,6 +1305,7 @@ TEST(tableMeta, updateStbMeta) {
tfree(tableMeta);
catalogDestroy();
memset(&gCtgMgmt.stat, 0, sizeof(gCtgMgmt.stat));
}
@ -1158,7 +1316,15 @@ TEST(tableDistVgroup, normalTable) {
SVgroupInfo *vgInfo = NULL;
SArray *vgList = NULL;
ctgTestSetPrepareDbVgroupsAndNormalMeta();
ctgTestInitLogFile();
memset(ctgTestRspFunc, 0, sizeof(ctgTestRspFunc));
ctgTestRspIdx = 0;
ctgTestRspFunc[0] = CTGT_RSP_VGINFO;
ctgTestRspFunc[1] = CTGT_RSP_TBMETA;
ctgTestRspFunc[2] = CTGT_RSP_VGINFO;
ctgTestSetRspByIdx();
initQueryModuleMsgHandle();
@ -1174,7 +1340,7 @@ TEST(tableDistVgroup, normalTable) {
strcpy(n.dbname, "db1");
strcpy(n.tname, ctgTestTablename);
code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
ASSERT_EQ(code, 0);
ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
vgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@ -1182,6 +1348,7 @@ TEST(tableDistVgroup, normalTable) {
ASSERT_EQ(vgInfo->epset.numOfEps, 3);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(tableDistVgroup, childTableCase) {
@ -1190,7 +1357,16 @@ TEST(tableDistVgroup, childTableCase) {
SVgroupInfo *vgInfo = NULL;
SArray *vgList = NULL;
ctgTestSetPrepareDbVgroupsAndChildMeta();
ctgTestInitLogFile();
memset(ctgTestRspFunc, 0, sizeof(ctgTestRspFunc));
ctgTestRspIdx = 0;
ctgTestRspFunc[0] = CTGT_RSP_VGINFO;
ctgTestRspFunc[1] = CTGT_RSP_CTBMETA;
ctgTestRspFunc[2] = CTGT_RSP_STBMETA;
ctgTestRspFunc[3] = CTGT_RSP_VGINFO;
ctgTestSetRspByIdx();
initQueryModuleMsgHandle();
@ -1206,7 +1382,7 @@ TEST(tableDistVgroup, childTableCase) {
strcpy(n.dbname, "db1");
strcpy(n.tname, ctgTestCTablename);
code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
ASSERT_EQ(code, 0);
ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
vgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@ -1214,6 +1390,7 @@ TEST(tableDistVgroup, childTableCase) {
ASSERT_EQ(vgInfo->epset.numOfEps, 4);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(tableDistVgroup, superTableCase) {
@ -1222,7 +1399,18 @@ TEST(tableDistVgroup, superTableCase) {
SVgroupInfo *vgInfo = NULL;
SArray *vgList = NULL;
ctgTestSetPrepareDbVgroupsAndSuperMeta();
ctgTestInitLogFile();
memset(ctgTestRspFunc, 0, sizeof(ctgTestRspFunc));
ctgTestRspIdx = 0;
ctgTestRspFunc[0] = CTGT_RSP_VGINFO;
ctgTestRspFunc[1] = CTGT_RSP_STBMETA;
ctgTestRspFunc[2] = CTGT_RSP_STBMETA;
ctgTestRspFunc[3] = CTGT_RSP_VGINFO;
ctgTestSetRspByIdx();
initQueryModuleMsgHandle();
@ -1237,7 +1425,7 @@ TEST(tableDistVgroup, superTableCase) {
strcpy(n.dbname, "db1");
strcpy(n.tname, ctgTestSTablename);
code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
ASSERT_EQ(code, 0);
ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 10);
vgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@ -1251,6 +1439,7 @@ TEST(tableDistVgroup, superTableCase) {
ASSERT_EQ(vgInfo->epset.numOfEps, 3);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(dbVgroup, getSetDbVgroupCase) {
@ -1258,12 +1447,19 @@ TEST(dbVgroup, getSetDbVgroupCase) {
void *mockPointer = (void *)0x1;
SVgroupInfo vgInfo = {0};
SVgroupInfo *pvgInfo = NULL;
SDBVgroupInfo *dbVgroup = NULL;
SDBVgInfo *dbVgroup = NULL;
SArray *vgList = NULL;
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroupsAndNormalMeta();
memset(ctgTestRspFunc, 0, sizeof(ctgTestRspFunc));
ctgTestRspIdx = 0;
ctgTestRspFunc[0] = CTGT_RSP_VGINFO;
ctgTestRspFunc[1] = CTGT_RSP_TBMETA;
ctgTestSetRspByIdx();
initQueryModuleMsgHandle();
@ -1279,16 +1475,21 @@ TEST(dbVgroup, getSetDbVgroupCase) {
strcpy(n.dbname, "db1");
strcpy(n.tname, ctgTestTablename);
code = catalogGetDBVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
code = catalogGetDBVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
ASSERT_EQ(code, 0);
ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), ctgTestVgNum);
while (0 == ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_DB_NUM)) {
usleep(10000);
}
code = catalogGetTableHashVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgInfo);
ASSERT_EQ(code, 0);
ASSERT_EQ(vgInfo.vgId, 8);
ASSERT_EQ(vgInfo.epset.numOfEps, 3);
code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
ASSERT_EQ(code, 0);
ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
pvgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@ -1297,15 +1498,26 @@ TEST(dbVgroup, getSetDbVgroupCase) {
taosArrayDestroy(vgList);
ctgTestBuildDBVgroup(&dbVgroup);
code = catalogUpdateDBVgroup(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
code = catalogUpdateDBVgInfo(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
ASSERT_EQ(code, 0);
while (true) {
uint64_t n = 0;
ctgDbgGetStatNum("runtime.qDoneNum", (void *)&n);
if (n != 3) {
usleep(10000);
} else {
break;
}
}
code = catalogGetTableHashVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgInfo);
ASSERT_EQ(code, 0);
ASSERT_EQ(vgInfo.vgId, 7);
ASSERT_EQ(vgInfo.epset.numOfEps, 2);
code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
ASSERT_EQ(code, 0);
ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
pvgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@ -1314,6 +1526,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
taosArrayDestroy(vgList);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(multiThread, getSetRmSameDbVgroup) {
@ -1321,13 +1534,13 @@ TEST(multiThread, getSetRmSameDbVgroup) {
void *mockPointer = (void *)0x1;
SVgroupInfo vgInfo = {0};
SVgroupInfo *pvgInfo = NULL;
SDBVgroupInfo dbVgroup = {0};
SDBVgInfo dbVgroup = {0};
SArray *vgList = NULL;
ctgTestStop = false;
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroups();
ctgTestSetRspDbVgroups();
initQueryModuleMsgHandle();
@ -1365,6 +1578,7 @@ TEST(multiThread, getSetRmSameDbVgroup) {
sleep(1);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
TEST(multiThread, getSetRmDiffDbVgroup) {
@ -1372,13 +1586,13 @@ TEST(multiThread, getSetRmDiffDbVgroup) {
void *mockPointer = (void *)0x1;
SVgroupInfo vgInfo = {0};
SVgroupInfo *pvgInfo = NULL;
SDBVgroupInfo dbVgroup = {0};
SDBVgInfo dbVgroup = {0};
SArray *vgList = NULL;
ctgTestStop = false;
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroups();
ctgTestSetRspDbVgroups();
initQueryModuleMsgHandle();
@ -1416,6 +1630,7 @@ TEST(multiThread, getSetRmDiffDbVgroup) {
sleep(1);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
@ -1425,13 +1640,13 @@ TEST(multiThread, ctableMeta) {
void *mockPointer = (void *)0x1;
SVgroupInfo vgInfo = {0};
SVgroupInfo *pvgInfo = NULL;
SDBVgroupInfo dbVgroup = {0};
SDBVgInfo dbVgroup = {0};
SArray *vgList = NULL;
ctgTestStop = false;
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroupsAndChildMeta();
ctgTestSetRspDbVgroupsAndChildMeta();
initQueryModuleMsgHandle();
@ -1468,6 +1683,7 @@ TEST(multiThread, ctableMeta) {
sleep(2);
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
@ -1477,7 +1693,7 @@ TEST(rentTest, allRent) {
void *mockPointer = (void *)0x1;
SVgroupInfo vgInfo = {0};
SVgroupInfo *pvgInfo = NULL;
SDBVgroupInfo dbVgroup = {0};
SDBVgInfo dbVgroup = {0};
SArray *vgList = NULL;
ctgTestStop = false;
SDbVgVersion *dbs = NULL;
@ -1486,7 +1702,7 @@ TEST(rentTest, allRent) {
ctgTestInitLogFile();
ctgTestSetPrepareDbVgroupsAndMultiSuperMeta();
ctgTestSetRspDbVgroupsAndMultiSuperMeta();
initQueryModuleMsgHandle();
@ -1516,6 +1732,10 @@ TEST(rentTest, allRent) {
ASSERT_EQ(tableMeta->tableInfo.precision, 1);
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
while (ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM) < i) {
usleep(10000);
}
code = catalogGetExpiredDBs(pCtg, &dbs, &num);
ASSERT_EQ(code, 0);
printf("%d - expired dbNum:%d\n", i, num);
@ -1541,6 +1761,7 @@ TEST(rentTest, allRent) {
}
catalogDestroy();
memset(&gCtgMgmt, 0, sizeof(gCtgMgmt));
}
int main(int argc, char **argv) {

2
source/libs/parser/src/astValidate.c
View File

@ -3647,7 +3647,7 @@ int32_t evaluateSqlNode(SSqlNode* pNode, int32_t tsPrecision, SMsgBuf* pMsgBuf)
//TODO remove it
int32_t setTableVgroupList(SParseContext *pCtx, SName* name, SVgroupsInfo **pVgList) {
SArray* vgroupList = NULL;
int32_t code = catalogGetTableDistVgroup(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, name, &vgroupList);
int32_t code = catalogGetTableDistVgInfo(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, name, &vgroupList);
if (code != TSDB_CODE_SUCCESS) {
return code;
}

2
source/libs/parser/src/dCDAstProcess.c
View File

@ -50,7 +50,7 @@ static int32_t setShowInfo(SShowInfo* pShowInfo, SParseContext* pCtx, void** out
char dbFname[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(&name, dbFname);
int32_t code = catalogGetDBVgroup(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, dbFname, false, &array);
int32_t code = catalogGetDBVgInfo(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, dbFname, false, &array);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
return code;

4
source/libs/qcom/src/querymsg.c
View File

@ -113,9 +113,9 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
}
pOut->dbVgroup = calloc(1, sizeof(SDBVgroupInfo));
pOut->dbVgroup = calloc(1, sizeof(SDBVgInfo));
if (NULL == pOut->dbVgroup) {
qError("calloc %d failed", (int32_t)sizeof(SDBVgroupInfo));
qError("calloc %d failed", (int32_t)sizeof(SDBVgInfo));
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}