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

add sma cache

This commit is contained in:
dapan1121 2022-06-13 10:38:45 +08:00
parent 31f3bed347
commit 17f6275015
13 changed files with 124 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
include/common/tmsg.h
View File

@ -1136,12 +1136,13 @@ int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp
void tFreeSTableMetaRsp(STableMetaRsp* pRsp); void tFreeSTableMetaRsp(STableMetaRsp* pRsp);
typedef struct { typedef struct {
SArray* pArray; // Array of STableMetaRsp SArray* pArray; // Array of STableMetaRsp
} STableMetaBatchRsp; STableIndexRsp* pIndexRsp;
} SSTbHbRsp;
int32_t tSerializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp); int32_t tSerializeSSTbHbRsp(void* buf, int32_t bufLen, SSTbHbRsp* pRsp);
int32_t tDeserializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp); int32_t tDeserializeSSTbHbRsp(void* buf, int32_t bufLen, SSTbHbRsp* pRsp);
void tFreeSTableMetaBatchRsp(STableMetaBatchRsp* pRsp); void tFreeSSTbHbRsp(SSTbHbRsp* pRsp);
typedef struct { typedef struct {
int32_t numOfTables; int32_t numOfTables;
@ -2513,8 +2514,9 @@ typedef struct {
} STableIndexInfo; } STableIndexInfo;
typedef struct { typedef struct {
int32_t version; uint64_t suid;
SArray* pIndex; int32_t version;
SArray* pIndex;
} STableIndexRsp; } STableIndexRsp;
int32_t tSerializeSTableIndexRsp(void* buf, int32_t bufLen, const STableIndexRsp* pRsp); int32_t tSerializeSTableIndexRsp(void* buf, int32_t bufLen, const STableIndexRsp* pRsp);

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

@ -98,14 +98,15 @@ typedef struct SCatalogCfg {
uint32_t stbRentSec; uint32_t stbRentSec;
} SCatalogCfg; } SCatalogCfg;
typedef struct SSTableMetaVersion { typedef struct SSTableVersion {
char dbFName[TSDB_DB_FNAME_LEN]; char dbFName[TSDB_DB_FNAME_LEN];
char stbName[TSDB_TABLE_NAME_LEN]; char stbName[TSDB_TABLE_NAME_LEN];
uint64_t dbId; uint64_t dbId;
uint64_t suid; uint64_t suid;
int16_t sversion; int16_t sversion;
int16_t tversion; int16_t tversion;
} SSTableMetaVersion; int32_t smaVer;
} SSTableVersion;
typedef struct SDbVgVersion { typedef struct SDbVgVersion {
char dbFName[TSDB_DB_FNAME_LEN]; char dbFName[TSDB_DB_FNAME_LEN];
@ -269,7 +270,7 @@ int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t
int32_t catalogGetQnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray* pQnodeList); int32_t catalogGetQnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray* pQnodeList);
int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num); int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableVersion **stables, uint32_t *num);
int32_t catalogGetExpiredDBs(SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num); int32_t catalogGetExpiredDBs(SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num);

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

@ -99,15 +99,15 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) { static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
int32_t code = 0; int32_t code = 0;
STableMetaBatchRsp batchMetaRsp = {0}; SSTbHbRsp hbRsp = {0};
if (tDeserializeSTableMetaBatchRsp(value, valueLen, &batchMetaRsp) != 0) { if (tDeserializeSSTbHbRsp(value, valueLen, &hbRsp) != 0) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
return -1; return -1;
} }
int32_t numOfBatchs = taosArrayGetSize(batchMetaRsp.pArray); int32_t numOfBatchs = taosArrayGetSize(hbRsp.pArray);
for (int32_t i = 0; i < numOfBatchs; ++i) { for (int32_t i = 0; i < numOfBatchs; ++i) {
STableMetaRsp *rsp = taosArrayGet(batchMetaRsp.pArray, i); STableMetaRsp *rsp = taosArrayGet(hbRsp.pArray, i);
if (rsp->numOfColumns < 0) { if (rsp->numOfColumns < 0) {
tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName); tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
@ -116,7 +116,7 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName); tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
if (rsp->pSchemas[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) { if (rsp->pSchemas[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
tscError("invalid colId[%" PRIi16 "] for the first column in table meta rsp msg", rsp->pSchemas[0].colId); tscError("invalid colId[%" PRIi16 "] for the first column in table meta rsp msg", rsp->pSchemas[0].colId);
tFreeSTableMetaBatchRsp(&batchMetaRsp); tFreeSSTbHbRsp(&hbRsp);
return TSDB_CODE_TSC_INVALID_VALUE; return TSDB_CODE_TSC_INVALID_VALUE;
} }
@ -124,7 +124,7 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
} }
} }
tFreeSTableMetaBatchRsp(&batchMetaRsp); tFreeSSTbHbRsp(&hbRsp);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -455,7 +455,7 @@ int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SCl
} }
int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) { int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
SSTableMetaVersion *stbs = NULL; SSTableVersion *stbs = NULL;
uint32_t stbNum = 0; uint32_t stbNum = 0;
int32_t code = 0; int32_t code = 0;
@ -469,15 +469,16 @@ int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SC
} }
for (int32_t i = 0; i < stbNum; ++i) { for (int32_t i = 0; i < stbNum; ++i) {
SSTableMetaVersion *stb = &stbs[i]; SSTableVersion *stb = &stbs[i];
stb->suid = htobe64(stb->suid); stb->suid = htobe64(stb->suid);
stb->sversion = htons(stb->sversion); stb->sversion = htons(stb->sversion);
stb->tversion = htons(stb->tversion); stb->tversion = htons(stb->tversion);
stb->smaVer = htonl(stb->smaVer);
} }
SKv kv = { SKv kv = {
.key = HEARTBEAT_KEY_STBINFO, .key = HEARTBEAT_KEY_STBINFO,
.valueLen = sizeof(SSTableMetaVersion) * stbNum, .valueLen = sizeof(SSTableVersion) * stbNum,
.value = stbs, .value = stbs,
}; };

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

@ -2437,6 +2437,7 @@ int32_t tSerializeSTableIndexRsp(void *buf, int32_t bufLen, const STableIndexRsp
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1; if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeU64(&encoder, pRsp->suid) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->version) < 0) return -1; if (tEncodeI32(&encoder, pRsp->version) < 0) return -1;
int32_t num = taosArrayGetSize(pRsp->pIndex); int32_t num = taosArrayGetSize(pRsp->pIndex);
if (tEncodeI32(&encoder, num) < 0) return -1; if (tEncodeI32(&encoder, num) < 0) return -1;
@ -2472,6 +2473,7 @@ int32_t tDeserializeSTableIndexRsp(void *buf, int32_t bufLen, STableIndexRsp *pR
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1; if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeU64(&decoder, &pRsp->suid) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->version) < 0) return -1; if (tDecodeI32(&decoder, &pRsp->version) < 0) return -1;
int32_t num = 0; int32_t num = 0;
if (tDecodeI32(&decoder, &num) < 0) return -1; if (tDecodeI32(&decoder, &num) < 0) return -1;
@ -2632,7 +2634,7 @@ int32_t tSerializeSTableMetaRsp(void *buf, int32_t bufLen, STableMetaRsp *pRsp)
return tlen; return tlen;
} }
int32_t tSerializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatchRsp *pRsp) { int32_t tSerializeSSTbHbRsp(void *buf, int32_t bufLen, SSTbHbRsp *pRsp) {
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
@ -2663,7 +2665,7 @@ int32_t tDeserializeSTableMetaRsp(void *buf, int32_t bufLen, STableMetaRsp *pRsp
return 0; return 0;
} }
int32_t tDeserializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatchRsp *pRsp) { int32_t tDeserializeSSTbHbRsp(void *buf, int32_t bufLen, SSTbHbRsp *pRsp) {
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
@ -2691,7 +2693,7 @@ int32_t tDeserializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatc
void tFreeSTableMetaRsp(STableMetaRsp *pRsp) { taosMemoryFreeClear(pRsp->pSchemas); } void tFreeSTableMetaRsp(STableMetaRsp *pRsp) { taosMemoryFreeClear(pRsp->pSchemas); }
void tFreeSTableMetaBatchRsp(STableMetaBatchRsp *pRsp) { void tFreeSSTbHbRsp(SSTbHbRsp *pRsp) {
int32_t numOfBatch = taosArrayGetSize(pRsp->pArray); int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
for (int32_t i = 0; i < numOfBatch; ++i) { for (int32_t i = 0; i < numOfBatch; ++i) {
STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pArray, i); STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pArray, i);

2
source/dnode/mnode/impl/inc/mndStb.h
View File

@ -27,7 +27,7 @@ void mndCleanupStb(SMnode *pMnode);
SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName); SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName);
void mndReleaseStb(SMnode *pMnode, SStbObj *pStb); void mndReleaseStb(SMnode *pMnode, SStbObj *pStb);
SSdbRaw *mndStbActionEncode(SStbObj *pStb); SSdbRaw *mndStbActionEncode(SStbObj *pStb);
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *pStbs, int32_t numOfStbs, void **ppRsp, int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbs, int32_t numOfStbs, void **ppRsp,
int32_t *pRspLen); int32_t *pRspLen);
int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs); int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs);

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

@ -433,7 +433,7 @@ static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHb
case HEARTBEAT_KEY_STBINFO: { case HEARTBEAT_KEY_STBINFO: {
void *rspMsg = NULL; void *rspMsg = NULL;
int32_t rspLen = 0; int32_t rspLen = 0;
mndValidateStbInfo(pMnode, kv->value, kv->valueLen / sizeof(SSTableMetaVersion), &rspMsg, &rspLen); mndValidateStbInfo(pMnode, kv->value, kv->valueLen / sizeof(SSTableVersion), &rspMsg, &rspLen);
if (rspMsg && rspLen > 0) { if (rspMsg && rspLen > 0) {
SKv kv1 = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = rspLen, .value = rspMsg}; SKv kv1 = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv1); taosArrayPush(hbRsp.info, &kv1);

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

@ -872,18 +872,29 @@ static int32_t mndGetSma(SMnode *pMnode, SUserIndexReq *indexReq, SUserIndexRsp
return code; return code;
} }
static int32_t mndGetTableSma(SMnode *pMnode, STableIndexReq *indexReq, STableIndexRsp *rsp, bool *exist) { static 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;
void *pIter = NULL; void *pIter = NULL;
STableIndexInfo info; STableIndexInfo info;
SStbObj* pStb = mndAcquireStb(pMnode, tbFName);
if (NULL == pStb) {
*exist = false;
return TSDB_CODE_SUCCESS;
}
rsp->suid = pStb->uid;
rsp->version = pStb->smaVer;
mndReleaseStb(pMnode, pStb);
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma); pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pSma->stb[0] != indexReq->tbFName[0] || strcmp(pSma->stb, indexReq->tbFName)) { if (pSma->stb[0] != tbFName[0] || strcmp(pSma->stb, tbFName)) {
continue; continue;
} }
@ -995,7 +1006,7 @@ static int32_t mndProcessGetTbSmaReq(SRpcMsg *pReq) {
goto _OVER; goto _OVER;
} }
code = mndGetTableSma(pMnode, &indexReq, &rsp, &exist); code = mndGetTableSma(pMnode, indexReq.tbFName, &rsp, &exist);
if (code) { if (code) {
goto _OVER; goto _OVER;
} }

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

@ -1271,7 +1271,7 @@ static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbNa
return 0; return 0;
} }
static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) { static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp, int32_t *smaVer) {
char tbFName[TSDB_TABLE_FNAME_LEN] = {0}; char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName); snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
@ -1288,6 +1288,10 @@ static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char
return -1; return -1;
} }
if (smaVer) {
*smaVer = pStb->smaVer;
}
int32_t code = mndBuildStbSchemaImp(pDb, pStb, tbName, pRsp); int32_t code = mndBuildStbSchemaImp(pDb, pStb, tbName, pRsp);
mndReleaseDb(pMnode, pDb); mndReleaseDb(pMnode, pDb);
mndReleaseStb(pMnode, pStb); mndReleaseStb(pMnode, pStb);
@ -1667,51 +1671,72 @@ _OVER:
return code; return code;
} }
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *pStbVersions, int32_t numOfStbs, void **ppRsp, int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
int32_t *pRspLen) { int32_t *pRspLen) {
STableMetaBatchRsp batchMetaRsp = {0}; SSTbHbRsp hbRsp = {0};
batchMetaRsp.pArray = taosArrayInit(numOfStbs, sizeof(STableMetaRsp)); hbRsp.pArray = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));
if (batchMetaRsp.pArray == NULL) { if (hbRsp.pArray == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
hbRsp.pIndexRsp = taosMemoryCalloc(1, sizeof(STableIndexRsp));
if (NULL == hbRsp.pIndexRsp) {
taosArrayDestroy(hbRsp.pArray);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < numOfStbs; ++i) { for (int32_t i = 0; i < numOfStbs; ++i) {
SSTableMetaVersion *pStbVersion = &pStbVersions[i]; SSTableVersion *pStbVersion = &pStbVersions[i];
pStbVersion->suid = be64toh(pStbVersion->suid); pStbVersion->suid = be64toh(pStbVersion->suid);
pStbVersion->sversion = ntohs(pStbVersion->sversion); pStbVersion->sversion = ntohs(pStbVersion->sversion);
pStbVersion->tversion = ntohs(pStbVersion->tversion); pStbVersion->tversion = ntohs(pStbVersion->tversion);
pStbVersion->smaVer = ntohl(pStbVersion->smaVer);
STableMetaRsp metaRsp = {0}; STableMetaRsp metaRsp = {0};
int32_t smaVer = 0;
mDebug("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName); mDebug("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName);
if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp) != 0) { if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp, &smaVer) != 0) {
metaRsp.numOfColumns = -1; metaRsp.numOfColumns = -1;
metaRsp.suid = pStbVersion->suid; metaRsp.suid = pStbVersion->suid;
taosArrayPush(hbRsp.pArray, &metaRsp);
continue;
} }
if (pStbVersion->sversion != metaRsp.sversion || pStbVersion->tversion != metaRsp.tversion) { if (pStbVersion->sversion != metaRsp.sversion || pStbVersion->tversion != metaRsp.tversion) {
taosArrayPush(batchMetaRsp.pArray, &metaRsp); taosArrayPush(hbRsp.pArray, &metaRsp);
} else { } else {
tFreeSTableMetaRsp(&metaRsp); tFreeSTableMetaRsp(&metaRsp);
} }
if (pStbVersion->smaVer != smaVer) {
bool exist = false;
char tbFName[TSDB_TABLE_FNAME_LEN];
sprintf(tbFName, "%s.%s", pStbVersion->dbFName, pStbVersion->stbName);
int32_t code = mndGetTableSma(pMnode, tbFName, hbRsp.pIndexRsp, &exist);
if (code || !exist) {
taosMemoryFreeClear(hbRsp.pIndexRsp);
}
}
} }
int32_t rspLen = tSerializeSTableMetaBatchRsp(NULL, 0, &batchMetaRsp); int32_t rspLen = tSerializeSSTbHbRsp(NULL, 0, &hbRsp);
if (rspLen < 0) { if (rspLen < 0) {
tFreeSTableMetaBatchRsp(&batchMetaRsp); tFreeSSTbHbRsp(&hbRsp);
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
return -1; return -1;
} }
void *pRsp = taosMemoryMalloc(rspLen); void *pRsp = taosMemoryMalloc(rspLen);
if (pRsp == NULL) { if (pRsp == NULL) {
tFreeSTableMetaBatchRsp(&batchMetaRsp); tFreeSSTbHbRsp(&hbRsp);
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
tSerializeSTableMetaBatchRsp(pRsp, rspLen, &batchMetaRsp); tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);
tFreeSTableMetaBatchRsp(&batchMetaRsp); tFreeSSTbHbRsp(&hbRsp);
*ppRsp = pRsp; *ppRsp = pRsp;
*pRspLen = rspLen; *pRspLen = rspLen;
return 0; return 0;

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

@ -156,7 +156,7 @@ typedef struct SCtgDBCache {
typedef struct SCtgRentSlot { typedef struct SCtgRentSlot {
SRWLatch lock; SRWLatch lock;
bool needSort; bool needSort;
SArray *meta; // element is SDbVgVersion or SSTableMetaVersion SArray *meta; // element is SDbVgVersion or SSTableVersion
} SCtgRentSlot; } SCtgRentSlot;
typedef struct SCtgRentMgmt { typedef struct SCtgRentMgmt {

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

@ -1074,14 +1074,14 @@ _return:
CTG_API_LEAVE(TSDB_CODE_SUCCESS); CTG_API_LEAVE(TSDB_CODE_SUCCESS);
} }
int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion **stables, uint32_t *num) { int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableVersion **stables, uint32_t *num) {
CTG_API_ENTER(); CTG_API_ENTER();
if (NULL == pCtg || NULL == stables || NULL == num) { if (NULL == pCtg || NULL == stables || NULL == num) {
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT); CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
} }
CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableMetaVersion))); CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableVersion)));
} }
int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) { int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) {

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

@ -970,7 +970,7 @@ int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t si
void *orig = taosArraySearch(slot->meta, &id, searchCompare, TD_EQ); void *orig = taosArraySearch(slot->meta, &id, searchCompare, TD_EQ);
if (NULL == orig) { if (NULL == orig) {
qError("meta not found in slot, id:%"PRIx64", slot idx:%d, type:%d, size:%d", id, widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta)); qDebug("meta not found in slot, id:%"PRIx64", slot idx:%d, type:%d, size:%d", id, widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR); CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
} }
@ -983,7 +983,7 @@ _return:
CTG_UNLOCK(CTG_WRITE, &slot->lock); CTG_UNLOCK(CTG_WRITE, &slot->lock);
if (code) { if (code) {
qWarn("meta in rent update failed, will try to add it, code:%x, id:%"PRIx64", slot idx:%d, type:%d", code, id, widx, mgmt->type); qDebug("meta in rent update failed, will try to add it, code:%x, id:%"PRIx64", slot idx:%d, type:%d", code, id, widx, mgmt->type);
CTG_RET(ctgMetaRentAdd(mgmt, meta, id, size)); CTG_RET(ctgMetaRentAdd(mgmt, meta, id, size));
} }
@ -1233,6 +1233,27 @@ int32_t ctgGetAddDBCache(SCatalog* pCtg, const char *dbFName, uint64_t dbId, SCt
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgUpdateRentStbVersion(SCatalog *pCtg, char* dbFName, char* tbName, uint64_t dbId, uint64_t suid, SCtgTbCache* pCache) {
SSTableVersion metaRent = {.dbId = dbId, .suid = suid};
if (pCache->pMeta) {
metaRent.sversion = pCache->pMeta->sversion;
metaRent.tversion = pCache->pMeta->tversion;
}
if (pCache->pIndex) {
metaRent.smaVer = pCache->pIndex->version;
}
strcpy(metaRent.dbFName, dbFName);
strcpy(metaRent.stbName, tbName);
CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->stbRent, &metaRent, metaRent.suid, sizeof(SSTableVersion), ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
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);
}
int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFName, uint64_t dbId, char *tbName, STableMeta *meta, int32_t metaSize) { int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFName, uint64_t dbId, char *tbName, STableMeta *meta, int32_t metaSize) {
if (NULL == dbCache->tbCache || NULL == dbCache->stbCache) { if (NULL == dbCache->tbCache || NULL == dbCache->stbCache) {
taosMemoryFree(meta); taosMemoryFree(meta);
@ -1263,7 +1284,6 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
ctgDebug("stb removed from stbCache, dbFName:%s, stb:%s, suid:%"PRIx64, dbFName, tbName, orig->suid); ctgDebug("stb removed from stbCache, dbFName:%s, stb:%s, suid:%"PRIx64, dbFName, tbName, orig->suid);
} }
ctgMetaRentRemove(&pCtg->stbRent, orig->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare);
origSuid = orig->suid; origSuid = orig->suid;
} }
} }
@ -1303,15 +1323,12 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType); ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
SSTableMetaVersion metaRent = {.dbId = dbId, .suid = meta->suid, .sversion = meta->sversion, .tversion = meta->tversion}; CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbId, meta->suid, pCache));
strcpy(metaRent.dbFName, dbFName);
strcpy(metaRent.stbName, tbName);
CTG_ERR_RET(ctgMetaRentAdd(&pCtg->stbRent, &metaRent, metaRent.suid, sizeof(SSTableMetaVersion)));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, 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) {
taosMemoryFreeClear(*index); taosMemoryFreeClear(*index);
ctgError("db is dropping, dbId:%"PRIx64, dbCache->dbId); ctgError("db is dropping, dbId:%"PRIx64, dbCache->dbId);
@ -1332,6 +1349,9 @@ int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *tbNam
*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, taosArrayGetSize(pIndex->pIndex));
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, pIndex->suid, pCache));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1344,6 +1364,8 @@ int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *tbNam
*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, taosArrayGetSize(pIndex->pIndex));
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, pIndex->suid, pCache));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1728,7 +1750,7 @@ int32_t ctgOpUpdateTbIndex(SCtgCacheOperation *operation) {
CTG_ERR_JRET(code); CTG_ERR_JRET(code);
} }
CTG_ERR_JRET(ctgWriteTbIndexToCache(pCtg, dbCache, msg->tbName, &pIndex)); CTG_ERR_JRET(ctgWriteTbIndexToCache(pCtg, dbCache, msg->dbFName, msg->tbName, &pIndex));
_return: _return:

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

@ -546,9 +546,9 @@ int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName
} }
int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2) { int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2) {
if (*(uint64_t *)key1 < ((SSTableMetaVersion*)key2)->suid) { if (*(uint64_t *)key1 < ((SSTableVersion*)key2)->suid) {
return -1; return -1;
} else if (*(uint64_t *)key1 > ((SSTableMetaVersion*)key2)->suid) { } else if (*(uint64_t *)key1 > ((SSTableVersion*)key2)->suid) {
return 1; return 1;
} else { } else {
return 0; return 0;
@ -566,9 +566,9 @@ int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2) {
} }
int32_t ctgStbVersionSortCompare(const void* key1, const void* key2) { int32_t ctgStbVersionSortCompare(const void* key1, const void* key2) {
if (((SSTableMetaVersion*)key1)->suid < ((SSTableMetaVersion*)key2)->suid) { if (((SSTableVersion*)key1)->suid < ((SSTableVersion*)key2)->suid) {
return -1; return -1;
} else if (((SSTableMetaVersion*)key1)->suid > ((SSTableMetaVersion*)key2)->suid) { } else if (((SSTableVersion*)key1)->suid > ((SSTableVersion*)key2)->suid) {
return 1; return 1;
} else { } else {
return 0; return 0;

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

@ -989,7 +989,7 @@ TEST(tableMeta, normalTable) {
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
SDbVgVersion *dbs = NULL; SDbVgVersion *dbs = NULL;
SSTableMetaVersion *stb = NULL; SSTableVersion *stb = NULL;
uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0; uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
int32_t i = 0; int32_t i = 0;
while (i < 5) { while (i < 5) {
@ -1098,7 +1098,7 @@ TEST(tableMeta, childTableCase) {
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
SDbVgVersion *dbs = NULL; SDbVgVersion *dbs = NULL;
SSTableMetaVersion *stb = NULL; SSTableVersion *stb = NULL;
uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0; uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
int32_t i = 0; int32_t i = 0;
while (i < 5) { while (i < 5) {
@ -1220,7 +1220,7 @@ TEST(tableMeta, superTableCase) {
ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
SDbVgVersion *dbs = NULL; SDbVgVersion *dbs = NULL;
SSTableMetaVersion *stb = NULL; SSTableVersion *stb = NULL;
uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0; uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
int32_t i = 0; int32_t i = 0;
while (i < 5) { while (i < 5) {
@ -2299,7 +2299,7 @@ TEST(rentTest, allRent) {
SArray *vgList = NULL; SArray *vgList = NULL;
ctgTestStop = false; ctgTestStop = false;
SDbVgVersion *dbs = NULL; SDbVgVersion *dbs = NULL;
SSTableMetaVersion *stable = NULL; SSTableVersion *stable = NULL;
uint32_t num = 0; uint32_t num = 0;
ctgTestInitLogFile(); ctgTestInitLogFile();