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

enh: refactor return code

This commit is contained in:
Hongze Cheng 2024-07-23 11:01:22 +08:00
parent 26c7943279
commit 10263b6b4e
9 changed files with 313 additions and 468 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/dnode/vnode/CMakeLists.txt
View File

@ -22,7 +22,6 @@ set(
# meta
"src/meta/metaOpen.c"
"src/meta/metaIdx.c"
"src/meta/metaTable.c"
"src/meta/metaSma.c"
"src/meta/metaQuery.c"

2
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -318,7 +318,7 @@ void* tdUidStoreFree(STbUidStore* pStore);
// SMetaSnapReader ========================================
int32_t metaSnapReaderOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapReader** ppReader);
int32_t metaSnapReaderClose(SMetaSnapReader** ppReader);
void metaSnapReaderClose(SMetaSnapReader** ppReader);
int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData);
// SMetaSnapWriter ========================================
int32_t metaSnapWriterOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapWriter** ppWriter);

133
source/dnode/vnode/src/meta/metaCache.c
View File

@ -118,90 +118,80 @@ static void freeCacheEntryFp(void* param) {
int32_t metaCacheOpen(SMeta* pMeta) {
int32_t code = 0;
SMetaCache* pCache = NULL;
int32_t lino;
pCache = (SMetaCache*)taosMemoryMalloc(sizeof(SMetaCache));
if (pCache == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
pMeta->pCache = (SMetaCache*)taosMemoryCalloc(1, sizeof(SMetaCache));
if (pMeta->pCache == NULL) {
TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
}
// open entry cache
pCache->sEntryCache.nEntry = 0;
pCache->sEntryCache.nBucket = META_CACHE_BASE_BUCKET;
pCache->sEntryCache.aBucket =
(SMetaCacheEntry**)taosMemoryCalloc(pCache->sEntryCache.nBucket, sizeof(SMetaCacheEntry*));
if (pCache->sEntryCache.aBucket == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
pMeta->pCache->sEntryCache.nEntry = 0;
pMeta->pCache->sEntryCache.nBucket = META_CACHE_BASE_BUCKET;
pMeta->pCache->sEntryCache.aBucket =
(SMetaCacheEntry**)taosMemoryCalloc(pMeta->pCache->sEntryCache.nBucket, sizeof(SMetaCacheEntry*));
if (pMeta->pCache->sEntryCache.aBucket == NULL) {
TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
}
// open stats cache
pCache->sStbStatsCache.nEntry = 0;
pCache->sStbStatsCache.nBucket = META_CACHE_STATS_BUCKET;
pCache->sStbStatsCache.aBucket =
(SMetaStbStatsEntry**)taosMemoryCalloc(pCache->sStbStatsCache.nBucket, sizeof(SMetaStbStatsEntry*));
if (pCache->sStbStatsCache.aBucket == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
pMeta->pCache->sStbStatsCache.nEntry = 0;
pMeta->pCache->sStbStatsCache.nBucket = META_CACHE_STATS_BUCKET;
pMeta->pCache->sStbStatsCache.aBucket =
(SMetaStbStatsEntry**)taosMemoryCalloc(pMeta->pCache->sStbStatsCache.nBucket, sizeof(SMetaStbStatsEntry*));
if (pMeta->pCache->sStbStatsCache.aBucket == NULL) {
TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
}
pCache->sTagFilterResCache.pUidResCache = taosLRUCacheInit(5 * 1024 * 1024, -1, 0.5);
if (pCache->sTagFilterResCache.pUidResCache == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
pMeta->pCache->sTagFilterResCache.pUidResCache = taosLRUCacheInit(5 * 1024 * 1024, -1, 0.5);
if (pMeta->pCache->sTagFilterResCache.pUidResCache == NULL) {
TSDB_CHECK_CODE(code = terrno, lino, _exit);
}
pCache->sTagFilterResCache.accTimes = 0;
pCache->sTagFilterResCache.pTableEntry =
pMeta->pCache->sTagFilterResCache.accTimes = 0;
pMeta->pCache->sTagFilterResCache.pTableEntry =
taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), false, HASH_NO_LOCK);
if (pCache->sTagFilterResCache.pTableEntry == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
if (pMeta->pCache->sTagFilterResCache.pTableEntry == NULL) {
TSDB_CHECK_CODE(code = terrno, lino, _exit);
}
taosHashSetFreeFp(pCache->sTagFilterResCache.pTableEntry, freeCacheEntryFp);
taosThreadMutexInit(&pCache->sTagFilterResCache.lock, NULL);
taosHashSetFreeFp(pMeta->pCache->sTagFilterResCache.pTableEntry, freeCacheEntryFp);
taosThreadMutexInit(&pMeta->pCache->sTagFilterResCache.lock, NULL);
pCache->STbGroupResCache.pResCache = taosLRUCacheInit(5 * 1024 * 1024, -1, 0.5);
if (pCache->STbGroupResCache.pResCache == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
pMeta->pCache->STbGroupResCache.pResCache = taosLRUCacheInit(5 * 1024 * 1024, -1, 0.5);
if (pMeta->pCache->STbGroupResCache.pResCache == NULL) {
TSDB_CHECK_CODE(code = terrno, lino, _exit);
}
pCache->STbGroupResCache.accTimes = 0;
pCache->STbGroupResCache.pTableEntry =
pMeta->pCache->STbGroupResCache.accTimes = 0;
pMeta->pCache->STbGroupResCache.pTableEntry =
taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), false, HASH_NO_LOCK);
if (pCache->STbGroupResCache.pTableEntry == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
if (pMeta->pCache->STbGroupResCache.pTableEntry == NULL) {
TSDB_CHECK_CODE(code = terrno, lino, _exit);
}
taosHashSetFreeFp(pCache->STbGroupResCache.pTableEntry, freeCacheEntryFp);
taosThreadMutexInit(&pCache->STbGroupResCache.lock, NULL);
taosHashSetFreeFp(pMeta->pCache->STbGroupResCache.pTableEntry, freeCacheEntryFp);
taosThreadMutexInit(&pMeta->pCache->STbGroupResCache.lock, NULL);
pCache->STbFilterCache.pStb = taosHashInit(0, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (pCache->STbFilterCache.pStb == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
pMeta->pCache->STbFilterCache.pStb =
taosHashInit(0, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (pMeta->pCache->STbFilterCache.pStb == NULL) {
TSDB_CHECK_CODE(code = terrno, lino, _exit);
}
pCache->STbFilterCache.pStbName =
pMeta->pCache->STbFilterCache.pStbName =
taosHashInit(0, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), false, HASH_NO_LOCK);
if (pCache->STbFilterCache.pStbName == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err2;
if (pMeta->pCache->STbFilterCache.pStbName == NULL) {
TSDB_CHECK_CODE(code = terrno, lino, _exit);
}
pMeta->pCache = pCache;
return code;
_err2:
entryCacheClose(pMeta);
_err:
taosMemoryFree(pCache);
metaError("vgId:%d, meta open cache failed since %s", TD_VID(pMeta->pVnode), tstrerror(code));
_exit:
if (code) {
metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));
metaCacheClose(pMeta);
} else {
metaDebug("vgId:%d, %s success", TD_VID(pMeta->pVnode), __func__);
}
return code;
}
@ -289,8 +279,7 @@ int32_t metaCacheUpsert(SMeta* pMeta, SMetaInfo* pInfo) {
}
} else { // insert
if (pCache->sEntryCache.nEntry >= pCache->sEntryCache.nBucket) {
code = metaRehashCache(pCache, 1);
if (code) goto _exit;
TAOS_UNUSED(metaRehashCache(pCache, 1));
iBucket = TABS(pInfo->uid) % pCache->sEntryCache.nBucket;
}
@ -328,8 +317,7 @@ int32_t metaCacheDrop(SMeta* pMeta, int64_t uid) {
pCache->sEntryCache.nEntry--;
if (pCache->sEntryCache.nEntry < pCache->sEntryCache.nBucket / 4 &&
pCache->sEntryCache.nBucket > META_CACHE_BASE_BUCKET) {
code = metaRehashCache(pCache, 0);
if (code) goto _exit;
TAOS_UNUSED(metaRehashCache(pCache, 0));
}
} else {
code = TSDB_CODE_NOT_FOUND;
@ -351,7 +339,9 @@ int32_t metaCacheGet(SMeta* pMeta, int64_t uid, SMetaInfo* pInfo) {
}
if (pEntry) {
if (pInfo) {
*pInfo = pEntry->info;
}
} else {
code = TSDB_CODE_NOT_FOUND;
}
@ -415,9 +405,7 @@ int32_t metaStatsCacheUpsert(SMeta* pMeta, SMetaStbStats* pInfo) {
(*ppEntry)->info.ctbNum = pInfo->ctbNum;
} else { // insert
if (pCache->sStbStatsCache.nEntry >= pCache->sStbStatsCache.nBucket) {
code = metaRehashStatsCache(pCache, 1);
if (code) goto _exit;
TAOS_UNUSED(metaRehashStatsCache(pCache, 1));
iBucket = TABS(pInfo->uid) % pCache->sStbStatsCache.nBucket;
}
@ -454,8 +442,7 @@ int32_t metaStatsCacheDrop(SMeta* pMeta, int64_t uid) {
pCache->sStbStatsCache.nEntry--;
if (pCache->sStbStatsCache.nEntry < pCache->sStbStatsCache.nBucket / 4 &&
pCache->sStbStatsCache.nBucket > META_CACHE_STATS_BUCKET) {
code = metaRehashStatsCache(pCache, 0);
if (code) goto _exit;
TAOS_UNUSED(metaRehashStatsCache(pCache, 0));
}
} else {
code = TSDB_CODE_NOT_FOUND;
@ -477,7 +464,9 @@ int32_t metaStatsCacheGet(SMeta* pMeta, int64_t uid, SMetaStbStats* pInfo) {
}
if (pEntry) {
if (pInfo) {
*pInfo = pEntry->info;
}
} else {
code = TSDB_CODE_NOT_FOUND;
}
@ -502,7 +491,9 @@ static int checkAllEntriesInCache(const STagFilterResEntry* pEntry, SArray* pInv
// check whether it is existed in LRU cache, and remove it from linked list if not.
LRUHandle* pRes = taosLRUCacheLookup(pCache, buf, len);
if (pRes == NULL) { // remove the item in the linked list
taosArrayPush(pInvalidRes, &pNode);
if (taosArrayPush(pInvalidRes, &pNode) == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
} else {
taosLRUCacheRelease(pCache, pRes, false);
}
@ -626,7 +617,7 @@ static int32_t addNewEntry(SHashObj* pTableEntry, const void* pKey, int32_t keyL
p->hitTimes = 0;
tdListInit(&p->list, keyLen);
taosHashPut(pTableEntry, &suid, sizeof(uint64_t), &p, POINTER_BYTES);
TAOS_CHECK_RETURN(taosHashPut(pTableEntry, &suid, sizeof(uint64_t), &p, POINTER_BYTES));
tdListAppend(&p->list, pKey);
return 0;
}
@ -956,9 +947,7 @@ int32_t metaInitTbFilterCache(SMeta* pMeta) {
}
if (tbNum && pTbArr) {
for (int32_t i = 0; i < tbNum; ++i) {
if (metaPutTbToFilterCache(pMeta, pTbArr[i], 1) != 0) {
return terrno ? terrno : -1;
}
TAOS_CHECK_RETURN(metaPutTbToFilterCache(pMeta, pTbArr[i], 1));
}
}
#else

42
source/dnode/vnode/src/meta/metaCommit.c
View File

@ -21,7 +21,10 @@ static FORCE_INLINE void *metaMalloc(void *pPool, size_t size) {
static FORCE_INLINE void metaFree(void *pPool, void *p) { vnodeBufPoolFree((SVBufPool *)pPool, p); }
// begin a meta txn
int metaBegin(SMeta *pMeta, int8_t heap) {
int32_t metaBegin(SMeta *pMeta, int8_t heap) {
int32_t code = 0;
int32_t lino;
void *(*xMalloc)(void *, size_t) = NULL;
void (*xFree)(void *, void *) = NULL;
void *xArg = NULL;
@ -36,12 +39,19 @@ int metaBegin(SMeta *pMeta, int8_t heap) {
xArg = pMeta->pVnode->inUse;
}
if (tdbBegin(pMeta->pEnv, &pMeta->txn, xMalloc, xFree, xArg, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
return -1;
code = tdbBegin(pMeta->pEnv, &pMeta->txn, xMalloc, xFree, xArg, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
TSDB_CHECK_CODE(code, lino, _exit);
code = tdbCommit(pMeta->pEnv, pMeta->txn);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
metaError("vgId:%d %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, __LINE__,
tstrerror(terrno));
} else {
metaDebug("vgId:%d %s success", TD_VID(pMeta->pVnode), __func__);
}
tdbCommit(pMeta->pEnv, pMeta->txn);
return 0;
}
@ -49,20 +59,36 @@ int metaBegin(SMeta *pMeta, int8_t heap) {
TXN *metaGetTxn(SMeta *pMeta) { return pMeta->txn; }
int metaCommit(SMeta *pMeta, TXN *txn) { return tdbCommit(pMeta->pEnv, txn); }
int metaFinishCommit(SMeta *pMeta, TXN *txn) { return tdbPostCommit(pMeta->pEnv, txn); }
int metaPrepareAsyncCommit(SMeta *pMeta) {
// return tdbPrepareAsyncCommit(pMeta->pEnv, pMeta->txn);
int code = 0;
int32_t lino;
metaWLock(pMeta);
code = ttlMgrFlush(pMeta->pTtlMgr, pMeta->txn);
TAOS_UNUSED(ttlMgrFlush(pMeta->pTtlMgr, pMeta->txn));
metaULock(pMeta);
code = tdbCommit(pMeta->pEnv, pMeta->txn);
TSDB_CHECK_CODE(code, lino, _exit);
pMeta->changed = false;
_exit:
if (code) {
metaError("vgId:%d %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, __LINE__,
tstrerror(terrno));
} else {
metaDebug("vgId:%d %s success", TD_VID(pMeta->pVnode), __func__);
}
return code;
}
// abort the meta txn
int metaAbort(SMeta *pMeta) {
if (!pMeta->txn) return 0;
if (!pMeta->txn) {
return 0;
}
int code = tdbAbort(pMeta->pEnv, pMeta->txn);
if (code) {
metaError("vgId:%d, failed to abort meta since %s", TD_VID(pMeta->pVnode), tstrerror(terrno));

143
source/dnode/vnode/src/meta/metaEntry.c
View File

@ -17,159 +17,166 @@
int meteEncodeColCmprEntry(SEncoder *pCoder, const SMetaEntry *pME) {
const SColCmprWrapper *pw = &pME->colCmpr;
if (tEncodeI32v(pCoder, pw->nCols) < 0) return -1;
if (tEncodeI32v(pCoder, pw->version) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeI32v(pCoder, pw->nCols));
TAOS_CHECK_RETURN(tEncodeI32v(pCoder, pw->version));
uDebug("encode cols:%d", pw->nCols);
for (int32_t i = 0; i < pw->nCols; i++) {
SColCmpr *p = &pw->pColCmpr[i];
if (tEncodeI16v(pCoder, p->id) < 0) return -1;
if (tEncodeU32(pCoder, p->alg) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeI16v(pCoder, p->id));
TAOS_CHECK_RETURN(tEncodeU32(pCoder, p->alg));
}
return 0;
}
int meteDecodeColCmprEntry(SDecoder *pDecoder, SMetaEntry *pME) {
SColCmprWrapper *pWrapper = &pME->colCmpr;
if (tDecodeI32v(pDecoder, &pWrapper->nCols) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeI32v(pDecoder, &pWrapper->nCols));
if (pWrapper->nCols == 0) {
return 0;
}
if (tDecodeI32v(pDecoder, &pWrapper->version) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeI32v(pDecoder, &pWrapper->version));
uDebug("dencode cols:%d", pWrapper->nCols);
pWrapper->pColCmpr = (SColCmpr *)tDecoderMalloc(pDecoder, pWrapper->nCols * sizeof(SColCmpr));
if (pWrapper->pColCmpr == NULL) return -1;
if (pWrapper->pColCmpr == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
for (int i = 0; i < pWrapper->nCols; i++) {
SColCmpr *p = &pWrapper->pColCmpr[i];
if (tDecodeI16v(pDecoder, &p->id) < 0) return -1;
if (tDecodeU32(pDecoder, &p->alg) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeI16v(pDecoder, &p->id));
TAOS_CHECK_RETURN(tDecodeU32(pDecoder, &p->alg));
}
return 0;
}
static FORCE_INLINE void metatInitDefaultSColCmprWrapper(SDecoder *pDecoder, SColCmprWrapper *pCmpr,
static FORCE_INLINE int32_t metatInitDefaultSColCmprWrapper(SDecoder *pDecoder, SColCmprWrapper *pCmpr,
SSchemaWrapper *pSchema) {
pCmpr->nCols = pSchema->nCols;
pCmpr->pColCmpr = (SColCmpr *)tDecoderMalloc(pDecoder, pCmpr->nCols * sizeof(SColCmpr));
if ((pCmpr->pColCmpr = (SColCmpr *)tDecoderMalloc(pDecoder, pCmpr->nCols * sizeof(SColCmpr))) == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
for (int32_t i = 0; i < pCmpr->nCols; i++) {
SColCmpr *pColCmpr = &pCmpr->pColCmpr[i];
SSchema *pColSchema = &pSchema->pSchema[i];
pColCmpr->id = pColSchema->colId;
pColCmpr->alg = createDefaultColCmprByType(pColSchema->type);
}
return 0;
}
int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) {
if (tStartEncode(pCoder) < 0) return -1;
TAOS_CHECK_RETURN(tStartEncode(pCoder));
TAOS_CHECK_RETURN(tEncodeI64(pCoder, pME->version));
TAOS_CHECK_RETURN(tEncodeI8(pCoder, pME->type));
TAOS_CHECK_RETURN(tEncodeI64(pCoder, pME->uid));
if (tEncodeI64(pCoder, pME->version) < 0) return -1;
if (tEncodeI8(pCoder, pME->type) < 0) return -1;
if (tEncodeI64(pCoder, pME->uid) < 0) return -1;
if (pME->name == NULL || tEncodeCStr(pCoder, pME->name) < 0) return -1;
if (pME->name == NULL) {
return TSDB_CODE_INVALID_PARA;
}
TAOS_CHECK_RETURN(tEncodeCStr(pCoder, pME->name));
if (pME->type == TSDB_SUPER_TABLE) {
if (tEncodeI8(pCoder, pME->flags) < 0) return -1; // TODO: need refactor?
if (tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaRow) < 0) return -1;
if (tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaTag) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeI8(pCoder, pME->flags));
TAOS_CHECK_RETURN(tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaRow));
TAOS_CHECK_RETURN(tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaTag));
if (TABLE_IS_ROLLUP(pME->flags)) {
if (tEncodeSRSmaParam(pCoder, &pME->stbEntry.rsmaParam) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeSRSmaParam(pCoder, &pME->stbEntry.rsmaParam));
}
} else if (pME->type == TSDB_CHILD_TABLE) {
if (tEncodeI64(pCoder, pME->ctbEntry.btime) < 0) return -1;
if (tEncodeI32(pCoder, pME->ctbEntry.ttlDays) < 0) return -1;
if (tEncodeI32v(pCoder, pME->ctbEntry.commentLen) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeI64(pCoder, pME->ctbEntry.btime));
TAOS_CHECK_RETURN(tEncodeI32(pCoder, pME->ctbEntry.ttlDays));
TAOS_CHECK_RETURN(tEncodeI32v(pCoder, pME->ctbEntry.commentLen));
if (pME->ctbEntry.commentLen > 0) {
if (tEncodeCStr(pCoder, pME->ctbEntry.comment) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeCStr(pCoder, pME->ctbEntry.comment));
}
if (tEncodeI64(pCoder, pME->ctbEntry.suid) < 0) return -1;
if (tEncodeTag(pCoder, (const STag *)pME->ctbEntry.pTags) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeI64(pCoder, pME->ctbEntry.suid));
TAOS_CHECK_RETURN(tEncodeTag(pCoder, (const STag *)pME->ctbEntry.pTags));
} else if (pME->type == TSDB_NORMAL_TABLE) {
if (tEncodeI64(pCoder, pME->ntbEntry.btime) < 0) return -1;
if (tEncodeI32(pCoder, pME->ntbEntry.ttlDays) < 0) return -1;
if (tEncodeI32v(pCoder, pME->ntbEntry.commentLen) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeI64(pCoder, pME->ntbEntry.btime));
TAOS_CHECK_RETURN(tEncodeI32(pCoder, pME->ntbEntry.ttlDays));
TAOS_CHECK_RETURN(tEncodeI32v(pCoder, pME->ntbEntry.commentLen));
if (pME->ntbEntry.commentLen > 0) {
if (tEncodeCStr(pCoder, pME->ntbEntry.comment) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeCStr(pCoder, pME->ntbEntry.comment));
}
if (tEncodeI32v(pCoder, pME->ntbEntry.ncid) < 0) return -1;
if (tEncodeSSchemaWrapper(pCoder, &pME->ntbEntry.schemaRow) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeI32v(pCoder, pME->ntbEntry.ncid));
TAOS_CHECK_RETURN(tEncodeSSchemaWrapper(pCoder, &pME->ntbEntry.schemaRow));
} else if (pME->type == TSDB_TSMA_TABLE) {
if (tEncodeTSma(pCoder, pME->smaEntry.tsma) < 0) return -1;
TAOS_CHECK_RETURN(tEncodeTSma(pCoder, pME->smaEntry.tsma));
} else {
metaError("meta/entry: invalide table type: %" PRId8 " encode failed.", pME->type);
return -1;
return TSDB_CODE_INVALID_PARA;
}
if (meteEncodeColCmprEntry(pCoder, pME) < 0) return -1;
TAOS_CHECK_RETURN(meteEncodeColCmprEntry(pCoder, pME));
tEndEncode(pCoder);
return 0;
}
int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
if (tStartDecode(pCoder) < 0) return -1;
if (tDecodeI64(pCoder, &pME->version) < 0) return -1;
if (tDecodeI8(pCoder, &pME->type) < 0) return -1;
if (tDecodeI64(pCoder, &pME->uid) < 0) return -1;
if (tDecodeCStr(pCoder, &pME->name) < 0) return -1;
TAOS_CHECK_RETURN(tStartDecode(pCoder));
TAOS_CHECK_RETURN(tDecodeI64(pCoder, &pME->version));
TAOS_CHECK_RETURN(tDecodeI8(pCoder, &pME->type));
TAOS_CHECK_RETURN(tDecodeI64(pCoder, &pME->uid));
TAOS_CHECK_RETURN(tDecodeCStr(pCoder, &pME->name));
if (pME->type == TSDB_SUPER_TABLE) {
if (tDecodeI8(pCoder, &pME->flags) < 0) return -1; // TODO: need refactor?
if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaRow) < 0) return -1;
if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaTag) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeI8(pCoder, &pME->flags));
TAOS_CHECK_RETURN(tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaRow));
TAOS_CHECK_RETURN(tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaTag));
if (TABLE_IS_ROLLUP(pME->flags)) {
if (tDecodeSRSmaParam(pCoder, &pME->stbEntry.rsmaParam) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeSRSmaParam(pCoder, &pME->stbEntry.rsmaParam));
}
} else if (pME->type == TSDB_CHILD_TABLE) {
if (tDecodeI64(pCoder, &pME->ctbEntry.btime) < 0) return -1;
if (tDecodeI32(pCoder, &pME->ctbEntry.ttlDays) < 0) return -1;
if (tDecodeI32v(pCoder, &pME->ctbEntry.commentLen) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeI64(pCoder, &pME->ctbEntry.btime));
TAOS_CHECK_RETURN(tDecodeI32(pCoder, &pME->ctbEntry.ttlDays));
TAOS_CHECK_RETURN(tDecodeI32v(pCoder, &pME->ctbEntry.commentLen));
if (pME->ctbEntry.commentLen > 0) {
if (tDecodeCStr(pCoder, &pME->ctbEntry.comment) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeCStr(pCoder, &pME->ctbEntry.comment));
}
if (tDecodeI64(pCoder, &pME->ctbEntry.suid) < 0) return -1;
if (tDecodeTag(pCoder, (STag **)&pME->ctbEntry.pTags) < 0) return -1; // (TODO)
TAOS_CHECK_RETURN(tDecodeI64(pCoder, &pME->ctbEntry.suid));
TAOS_CHECK_RETURN(tDecodeTag(pCoder, (STag **)&pME->ctbEntry.pTags));
} else if (pME->type == TSDB_NORMAL_TABLE) {
if (tDecodeI64(pCoder, &pME->ntbEntry.btime) < 0) return -1;
if (tDecodeI32(pCoder, &pME->ntbEntry.ttlDays) < 0) return -1;
if (tDecodeI32v(pCoder, &pME->ntbEntry.commentLen) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeI64(pCoder, &pME->ntbEntry.btime));
TAOS_CHECK_RETURN(tDecodeI32(pCoder, &pME->ntbEntry.ttlDays));
TAOS_CHECK_RETURN(tDecodeI32v(pCoder, &pME->ntbEntry.commentLen));
if (pME->ntbEntry.commentLen > 0) {
if (tDecodeCStr(pCoder, &pME->ntbEntry.comment) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeCStr(pCoder, &pME->ntbEntry.comment));
}
if (tDecodeI32v(pCoder, &pME->ntbEntry.ncid) < 0) return -1;
if (tDecodeSSchemaWrapperEx(pCoder, &pME->ntbEntry.schemaRow) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeI32v(pCoder, &pME->ntbEntry.ncid));
TAOS_CHECK_RETURN(tDecodeSSchemaWrapperEx(pCoder, &pME->ntbEntry.schemaRow));
} else if (pME->type == TSDB_TSMA_TABLE) {
pME->smaEntry.tsma = tDecoderMalloc(pCoder, sizeof(STSma));
if (!pME->smaEntry.tsma) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
return TSDB_CODE_OUT_OF_MEMORY;
}
if (tDecodeTSma(pCoder, pME->smaEntry.tsma, true) < 0) return -1;
TAOS_CHECK_RETURN(tDecodeTSma(pCoder, pME->smaEntry.tsma, true));
} else {
metaError("meta/entry: invalide table type: %" PRId8 " decode failed.", pME->type);
return -1;
return TSDB_CODE_INVALID_PARA;
}
if (pME->type == TSDB_SUPER_TABLE) {
if (TABLE_IS_COL_COMPRESSED(pME->flags)) {
if (meteDecodeColCmprEntry(pCoder, pME) < 0) return -1;
TAOS_CHECK_RETURN(meteDecodeColCmprEntry(pCoder, pME));
if (pME->colCmpr.nCols == 0) {
metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->stbEntry.schemaRow);
TAOS_CHECK_RETURN(metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->stbEntry.schemaRow));
}
} else {
metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->stbEntry.schemaRow);
TAOS_CHECK_RETURN(metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->stbEntry.schemaRow));
TABLE_SET_COL_COMPRESSED(pME->flags);
}
} else if (pME->type == TSDB_NORMAL_TABLE) {
if (!tDecodeIsEnd(pCoder)) {
uDebug("set type: %d, tableName:%s", pME->type, pME->name);
if (meteDecodeColCmprEntry(pCoder, pME) < 0) return -1;
TAOS_CHECK_RETURN(meteDecodeColCmprEntry(pCoder, pME));
if (pME->colCmpr.nCols == 0) {
metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->ntbEntry.schemaRow);
TAOS_CHECK_RETURN(metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->ntbEntry.schemaRow));
}
} else {
uDebug("set default type: %d, tableName:%s", pME->type, pME->name);
metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->ntbEntry.schemaRow);
TAOS_CHECK_RETURN(metatInitDefaultSColCmprWrapper(pCoder, &pME->colCmpr, &pME->ntbEntry.schemaRow));
}
TABLE_SET_COL_COMPRESSED(pME->flags);
}

118
source/dnode/vnode/src/meta/metaIdx.c
View File

@ -1,118 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifdef USE_INVERTED_INDEX
#include "index.h"
#endif
#include "meta.h"
struct SMetaIdx {
#ifdef USE_INVERTED_INDEX
SIndex *pIdx;
#endif
/* data */
#ifdef WINDOWS
size_t avoidCompilationErrors;
#endif
};
int metaOpenIdx(SMeta *pMeta) {
#if 0
char idxDir[128]; // TODO
char * err = NULL;
rocksdb_options_t *options = rocksdb_options_create();
// TODO
sprintf(idxDir, "%s/index", pMeta->path);
if (pMeta->pCache) {
rocksdb_options_set_row_cache(options, pMeta->pCache);
}
rocksdb_options_set_create_if_missing(options, 1);
pMeta->pIdx = rocksdb_open(options, idxDir, &err);
if (pMeta->pIdx == NULL) {
// TODO: handle error
rocksdb_options_destroy(options);
return -1;
}
rocksdb_options_destroy(options);
#endif
#ifdef USE_INVERTED_INDEX
// SIndexOpts opts;
// if (indexOpen(&opts, pMeta->path, &pMeta->pIdx->pIdx) != 0) {
// return -1;
//}
#endif
return 0;
}
#ifdef BUILD_NO_CALL
void metaCloseIdx(SMeta *pMeta) { /* TODO */
#if 0
if (pMeta->pIdx) {
rocksdb_close(pMeta->pIdx);
pMeta->pIdx = NULL;
}
#endif
#ifdef USE_INVERTED_INDEX
// SIndexOpts opts;
// if (indexClose(pMeta->pIdx->pIdx) != 0) {
// return -1;
//}
// return 0;
#endif
}
int metaSaveTableToIdx(SMeta *pMeta, const STbCfg *pTbCfg) {
#ifdef USE_INVERTED_INDEX
// if (pTbCfgs->type == META_CHILD_TABLE) {
// char buf[8] = {0};
// int16_t colId = (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId;
// sprintf(buf, "%d", colId); // colname
// char *pTagVal = (char *)tdGetKVRowValOfCol(pTbCfg->ctbCfg.pTag, (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId);
// tb_uid_t suid = pTbCfg->ctbCfg.suid; // super id
// tb_uid_t tuid = 0; // child table uid
// SIndexMultiTerm *terms = indexMultiTermCreate();
// SIndexTerm *term =
// indexTermCreate(suid, ADD_VALUE, TSDB_DATA_TYPE_BINARY, buf, strlen(buf), pTagVal, strlen(pTagVal), tuid);
// indexMultiTermAdd(terms, term);
// int ret = indexPut(pMeta->pIdx->pIdx, terms);
// indexMultiTermDestroy(terms);
// return ret;
//} else {
// return DB_DONOTINDEX;
//}
#endif
// TODO
return 0;
}
int metaRemoveTableFromIdx(SMeta *pMeta, tb_uid_t uid) {
#ifdef USE_INVERTED_INDEX
#endif
// TODO
return 0;
}
#endif

211
source/dnode/vnode/src/meta/metaOpen.c
View File

@ -39,13 +39,14 @@ static int32_t metaDestroyLock(SMeta *pMeta) { return taosThreadRwlockDestroy(&p
static void metaCleanup(SMeta **ppMeta);
int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
int32_t metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
SMeta *pMeta = NULL;
int ret;
int offset;
int32_t code = 0;
int32_t lino;
int32_t offset;
char path[TSDB_FILENAME_LEN] = {0};
*ppMeta = NULL;
char indexFullPath[128] = {0};
sprintf(indexFullPath, "%s/%s", pMeta->path, "invert");
// create handle
vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pVnode->pTfs, path, TSDB_FILENAME_LEN);
@ -53,8 +54,7 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VNODE_META_DIR);
if ((pMeta = taosMemoryCalloc(1, sizeof(*pMeta) + strlen(path) + 1)) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
}
metaInitLock(pMeta);
@ -69,163 +69,103 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
taosMkDir(pMeta->path);
// open env
ret = tdbOpen(pMeta->path, pVnode->config.szPage, pVnode->config.szCache, &pMeta->pEnv, rollback,
code = tdbOpen(pMeta->path, pVnode->config.szPage, pVnode->config.szCache, &pMeta->pEnv, rollback,
pVnode->config.tdbEncryptAlgorithm, pVnode->config.tdbEncryptKey);
if (ret < 0) {
metaError("vgId:%d, failed to open meta env since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
TSDB_CHECK_CODE(code, lino, _exit);
// open pTbDb
ret = tdbTbOpen("table.db", sizeof(STbDbKey), -1, tbDbKeyCmpr, pMeta->pEnv, &pMeta->pTbDb, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta table db since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("table.db", sizeof(STbDbKey), -1, tbDbKeyCmpr, pMeta->pEnv, &pMeta->pTbDb, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// open pSkmDb
ret = tdbTbOpen("schema.db", sizeof(SSkmDbKey), -1, skmDbKeyCmpr, pMeta->pEnv, &pMeta->pSkmDb, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta schema db since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("schema.db", sizeof(SSkmDbKey), -1, skmDbKeyCmpr, pMeta->pEnv, &pMeta->pSkmDb, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// open pUidIdx
ret = tdbTbOpen("uid.idx", sizeof(tb_uid_t), sizeof(SUidIdxVal), uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pUidIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta uid idx since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("uid.idx", sizeof(tb_uid_t), sizeof(SUidIdxVal), uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pUidIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// open pNameIdx
ret = tdbTbOpen("name.idx", -1, sizeof(tb_uid_t), NULL, pMeta->pEnv, &pMeta->pNameIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta name index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("name.idx", -1, sizeof(tb_uid_t), NULL, pMeta->pEnv, &pMeta->pNameIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// open pCtbIdx
ret = tdbTbOpen("ctb.idx", sizeof(SCtbIdxKey), -1, ctbIdxKeyCmpr, pMeta->pEnv, &pMeta->pCtbIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta child table index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("ctb.idx", sizeof(SCtbIdxKey), -1, ctbIdxKeyCmpr, pMeta->pEnv, &pMeta->pCtbIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// open pSuidIdx
ret = tdbTbOpen("suid.idx", sizeof(tb_uid_t), 0, uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pSuidIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta super table index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("suid.idx", sizeof(tb_uid_t), 0, uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pSuidIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
char indexFullPath[128] = {0};
sprintf(indexFullPath, "%s/%s", pMeta->path, "invert");
taosMkDir(indexFullPath);
TAOS_UNUSED(taosMkDir(indexFullPath));
SIndexOpts opts = {.cacheSize = 8 * 1024 * 1024};
ret = indexOpen(&opts, indexFullPath, (SIndex **)&pMeta->pTagIvtIdx);
if (ret < 0) {
metaError("vgId:%d, failed to open meta tag index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = indexOpen(&opts, indexFullPath, (SIndex **)&pMeta->pTagIvtIdx);
TSDB_CHECK_CODE(code, lino, _exit);
ret = tdbTbOpen("tag.idx", -1, 0, tagIdxKeyCmpr, pMeta->pEnv, &pMeta->pTagIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta tag index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("tag.idx", -1, 0, tagIdxKeyCmpr, pMeta->pEnv, &pMeta->pTagIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// open pTtlMgr ("ttlv1.idx")
char logPrefix[128] = {0};
sprintf(logPrefix, "vgId:%d", TD_VID(pVnode));
ret = ttlMgrOpen(&pMeta->pTtlMgr, pMeta->pEnv, 0, logPrefix, tsTtlFlushThreshold);
if (ret < 0) {
metaError("vgId:%d, failed to open meta ttl index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = ttlMgrOpen(&pMeta->pTtlMgr, pMeta->pEnv, 0, logPrefix, tsTtlFlushThreshold);
TSDB_CHECK_CODE(code, lino, _exit);
// open pSmaIdx
ret = tdbTbOpen("sma.idx", sizeof(SSmaIdxKey), 0, smaIdxKeyCmpr, pMeta->pEnv, &pMeta->pSmaIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta sma index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("sma.idx", sizeof(SSmaIdxKey), 0, smaIdxKeyCmpr, pMeta->pEnv, &pMeta->pSmaIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// idx table create time
ret = tdbTbOpen("ctime.idx", sizeof(SBtimeIdxKey), 0, btimeIdxCmpr, pMeta->pEnv, &pMeta->pBtimeIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta ctime index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("ctime.idx", sizeof(SBtimeIdxKey), 0, btimeIdxCmpr, pMeta->pEnv, &pMeta->pBtimeIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// idx num of col, normal table only
ret = tdbTbOpen("ncol.idx", sizeof(SNcolIdxKey), 0, ncolIdxCmpr, pMeta->pEnv, &pMeta->pNcolIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta ncol index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("ncol.idx", sizeof(SNcolIdxKey), 0, ncolIdxCmpr, pMeta->pEnv, &pMeta->pNcolIdx, 0);
TSDB_CHECK_CODE(code, lino, _exit);
ret = tdbTbOpen("stream.task.db", sizeof(int64_t), -1, taskIdxKeyCmpr, pMeta->pEnv, &pMeta->pStreamDb, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta stream task index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = tdbTbOpen("stream.task.db", sizeof(int64_t), -1, taskIdxKeyCmpr, pMeta->pEnv, &pMeta->pStreamDb, 0);
TSDB_CHECK_CODE(code, lino, _exit);
// open index
if (metaOpenIdx(pMeta) < 0) {
metaError("vgId:%d, failed to open meta index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
code = metaCacheOpen(pMeta);
TSDB_CHECK_CODE(code, lino, _exit);
int32_t code = metaCacheOpen(pMeta);
code = metaInitTbFilterCache(pMeta);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
terrno = code;
metaError("vgId:%d, failed to open meta cache since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
if (metaInitTbFilterCache(pMeta) != 0) {
goto _err;
}
metaDebug("vgId:%d, meta is opened", TD_VID(pVnode));
*ppMeta = pMeta;
return 0;
_err:
metaError("vgId:%d %s failed at %s:%d since %s", TD_VID(pVnode), __func__, __FILE__, __LINE__, tstrerror(code));
metaCleanup(&pMeta);
return -1;
*ppMeta = NULL;
} else {
metaDebug("vgId:%d %s success", TD_VID(pVnode), __func__);
*ppMeta = pMeta;
}
return code;
}
int metaUpgrade(SVnode *pVnode, SMeta **ppMeta) {
int code = TSDB_CODE_SUCCESS;
int32_t metaUpgrade(SVnode *pVnode, SMeta **ppMeta) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino;
SMeta *pMeta = *ppMeta;
if (ttlMgrNeedUpgrade(pMeta->pEnv)) {
code = metaBegin(pMeta, META_BEGIN_HEAP_OS);
if (code < 0) {
metaError("vgId:%d, failed to upgrade meta, meta begin failed since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
TSDB_CHECK_CODE(code, lino, _exit);
code = ttlMgrUpgrade(pMeta->pTtlMgr, pMeta);
if (code < 0) {
metaError("vgId:%d, failed to upgrade meta ttl since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
TSDB_CHECK_CODE(code, lino, _exit);
code = metaCommit(pMeta, pMeta->txn);
if (code < 0) {
metaError("vgId:%d, failed to upgrade meta ttl, meta commit failed since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
TSDB_CHECK_CODE(code, lino, _exit);
}
return TSDB_CODE_SUCCESS;
_err:
_exit:
if (code) {
metaError("vgId:%d %s failed at %s:%d since %s", TD_VID(pVnode), __func__, __FILE__, __LINE__, tstrerror(code));
metaCleanup(ppMeta);
}
return code;
}
@ -235,33 +175,42 @@ int metaClose(SMeta **ppMeta) {
}
int metaAlterCache(SMeta *pMeta, int32_t nPage) {
int32_t code = 0;
metaWLock(pMeta);
if (tdbAlter(pMeta->pEnv, nPage) < 0) {
code = tdbAlter(pMeta->pEnv, nPage);
metaULock(pMeta);
return -1;
if (code) {
metaError("vgId:%d %s failed since %s", TD_VID(pMeta->pVnode), __func__, tstrerror(code));
}
metaULock(pMeta);
return 0;
return code;
}
int32_t metaRLock(SMeta *pMeta) {
metaTrace("meta rlock %p", &pMeta->lock);
int32_t ret = taosThreadRwlockRdlock(&pMeta->lock);
return ret;
int32_t code = taosThreadRwlockRdlock(&pMeta->lock);
if (code) {
return TAOS_SYSTEM_ERROR(code);
}
return 0;
}
int32_t metaWLock(SMeta *pMeta) {
metaTrace("meta wlock %p", &pMeta->lock);
int32_t ret = taosThreadRwlockWrlock(&pMeta->lock);
return ret;
int32_t code = taosThreadRwlockWrlock(&pMeta->lock);
if (code) {
return TAOS_SYSTEM_ERROR(code);
}
return 0;
}
int32_t metaULock(SMeta *pMeta) {
metaTrace("meta ulock %p", &pMeta->lock);
int32_t ret = taosThreadRwlockUnlock(&pMeta->lock);
return ret;
int32_t code = taosThreadRwlockUnlock(&pMeta->lock);
if (code) {
return TAOS_SYSTEM_ERROR(code);
}
return 0;
}
static void metaCleanup(SMeta **ppMeta) {

92
source/dnode/vnode/src/meta/metaSnapshot.c
View File

@ -25,14 +25,14 @@ struct SMetaSnapReader {
int32_t metaSnapReaderOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapReader** ppReader) {
int32_t code = 0;
int32_t lino;
int32_t c = 0;
SMetaSnapReader* pReader = NULL;
// alloc
pReader = (SMetaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));
if (pReader == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
}
pReader->pMeta = pMeta;
pReader->sver = sver;
@ -40,36 +40,29 @@ int32_t metaSnapReaderOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapRe
// impl
code = tdbTbcOpen(pMeta->pTbDb, &pReader->pTbc, NULL);
if (code) {
taosMemoryFree(pReader);
goto _err;
}
TSDB_CHECK_CODE(code, lino, _exit);
code = tdbTbcMoveTo(pReader->pTbc, &(STbDbKey){.version = sver, .uid = INT64_MIN}, sizeof(STbDbKey), &c);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
taosMemoryFree(pReader);
goto _err;
}
metaInfo("vgId:%d, vnode snapshot meta reader opened", TD_VID(pMeta->pVnode));
*ppReader = pReader;
return code;
_err:
metaError("vgId:%d, vnode snapshot meta reader open failed since %s", TD_VID(pMeta->pVnode), tstrerror(code));
metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));
metaSnapReaderClose(&pReader);
*ppReader = NULL;
} else {
metaInfo("vgId:%d, %s success", TD_VID(pMeta->pVnode), __func__);
*ppReader = pReader;
}
return code;
}
int32_t metaSnapReaderClose(SMetaSnapReader** ppReader) {
int32_t code = 0;
void metaSnapReaderClose(SMetaSnapReader** ppReader) {
if (ppReader && *ppReader) {
tdbTbcClose((*ppReader)->pTbc);
taosMemoryFree(*ppReader);
*ppReader = NULL;
return code;
}
}
int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {
@ -106,7 +99,7 @@ int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {
*ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + nData);
if (*ppData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
goto _exit;
}
SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
@ -122,10 +115,10 @@ int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {
}
_exit:
return code;
_err:
metaError("vgId:%d, vnode snapshot meta read data failed since %s", TD_VID(pReader->pMeta->pVnode), tstrerror(code));
if (code) {
metaError("vgId:%d, vnode snapshot meta read data failed since %s", TD_VID(pReader->pMeta->pVnode),
tstrerror(code));
}
return code;
}
@ -138,26 +131,30 @@ struct SMetaSnapWriter {
int32_t metaSnapWriterOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapWriter** ppWriter) {
int32_t code = 0;
int32_t lino;
SMetaSnapWriter* pWriter;
// alloc
pWriter = (SMetaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
if (pWriter == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
}
pWriter->pMeta = pMeta;
pWriter->sver = sver;
pWriter->ever = ever;
metaBegin(pMeta, META_BEGIN_HEAP_NIL);
code = metaBegin(pMeta, META_BEGIN_HEAP_NIL);
TSDB_CHECK_CODE(code, lino, _exit);
*ppWriter = pWriter;
return code;
_err:
metaError("vgId:%d, meta snapshot writer open failed since %s", TD_VID(pMeta->pVnode), tstrerror(code));
_exit:
if (code) {
metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));
taosMemoryFree(pWriter);
*ppWriter = NULL;
} else {
metaDebug("vgId:%d, %s success", TD_VID(pMeta->pVnode), __func__);
*ppWriter = pWriter;
}
return code;
}
@ -189,25 +186,24 @@ _err:
int32_t metaSnapWrite(SMetaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
int32_t code = 0;
int32_t line = 0;
int32_t lino = 0;
SMeta* pMeta = pWriter->pMeta;
SMetaEntry metaEntry = {0};
SDecoder* pDecoder = &(SDecoder){0};
tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
code = metaDecodeEntry(pDecoder, &metaEntry);
VND_CHECK_CODE(code, line, _err);
TSDB_CHECK_CODE(code, lino, _exit);
code = metaHandleEntry(pMeta, &metaEntry);
VND_CHECK_CODE(code, line, _err);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));
}
tDecoderClear(pDecoder);
return code;
_err:
tDecoderClear(pDecoder);
metaError("vgId:%d, vnode snapshot meta write failed since %s at line:%d", TD_VID(pMeta->pVnode), terrstr(), line);
return code;
}
typedef struct STableInfoForChildTable {
@ -468,19 +464,17 @@ int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {
return c;
}
void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid){
void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {
bool ret = false;
SSchemaWrapper *schema = metaGetTableSchema(ctx->pMeta, uid, -1, 1);
if (schema->nCols >= 2 && schema->pSchema[1].flags & COL_IS_KEY){
SSchemaWrapper* schema = metaGetTableSchema(ctx->pMeta, uid, -1, 1);
if (schema->nCols >= 2 && schema->pSchema[1].flags & COL_IS_KEY) {
ret = true;
}
tDeleteSchemaWrapper(schema);
ctx->hasPrimaryKey = ret;
}
bool taosXGetTablePrimaryKey(SSnapContext* ctx){
return ctx->hasPrimaryKey;
}
bool taosXGetTablePrimaryKey(SSnapContext* ctx) { return ctx->hasPrimaryKey; }
int32_t getTableInfoFromSnapshot(SSnapContext* ctx, void** pBuf, int32_t* contLen, int16_t* type, int64_t* uid) {
int32_t ret = 0;

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

@ -312,8 +312,7 @@ int32_t vnodeSnapRead(SVSnapReader *pReader, uint8_t **ppData, uint32_t *nData)
goto _exit;
} else {
pReader->metaDone = 1;
code = metaSnapReaderClose(&pReader->pMetaReader);
TSDB_CHECK_CODE(code, lino, _exit);
metaSnapReaderClose(&pReader->pMetaReader);
}
}