diff --git a/source/dnode/vnode/src/inc/tsdb.h b/source/dnode/vnode/src/inc/tsdb.h index ed8f99ec75..c1123db7a3 100644 --- a/source/dnode/vnode/src/inc/tsdb.h +++ b/source/dnode/vnode/src/inc/tsdb.h @@ -342,7 +342,10 @@ typedef struct { rocksdb_writeoptions_t *writeoptions; rocksdb_readoptions_t *readoptions; rocksdb_writebatch_t *writebatch; - TdThreadMutex writeBatchMutex; + TdThreadMutex writeBatchMutex; + int32_t sver; + tb_uid_t suid; + tb_uid_t uid; STSchema *pTSchema; } SRocksCache; diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 54efa0314b..512e088428 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -222,6 +222,7 @@ int32_t tsdbCacheNewSTableColumn(STsdb* pTsdb, SArray* uids, int16_t cid, int8_t int32_t tsdbCacheDropSTableColumn(STsdb* pTsdb, SArray* uids, int16_t cid, bool hasPrimayKey); int32_t tsdbCacheNewNTableColumn(STsdb* pTsdb, int64_t uid, int16_t cid, int8_t col_type); int32_t tsdbCacheDropNTableColumn(STsdb* pTsdb, int64_t uid, int16_t cid, bool hasPrimayKey); +void tsdbCacheInvalidateSchema(STsdb* pTsdb, tb_uid_t suid, tb_uid_t uid, int32_t sver); int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq2* pMsg); int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq2* pMsg, SSubmitRsp2* pRsp); int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitTbData* pSubmitTbData, int32_t* affectedRows); diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index 21d12ef77d..5c3516a962 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -620,6 +620,8 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { } } if (uids) taosArrayDestroy(uids); + + tsdbCacheInvalidateSchema(pTsdb, pReq->suid, -1, pReq->schemaRow.version); } metaWLock(pMeta); @@ -1945,6 +1947,10 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl break; } + if (!TSDB_CACHE_NO(pMeta->pVnode->config)) { + tsdbCacheInvalidateSchema(pMeta->pVnode->pTsdb, 0, entry.uid, pSchema->version); + } + entry.version = version; // do actual write diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c index cbb4f9e873..2cef541cdb 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCache.c +++ b/source/dnode/vnode/src/tsdb/tsdbCache.c @@ -221,7 +221,7 @@ static int32_t tsdbOpenRocksCache(STsdb *pTsdb) { rocksdb_writebatch_t *writebatch = rocksdb_writebatch_create(); - TAOS_CHECK_GOTO(taosThreadMutexInit(&pTsdb->rCache.writeBatchMutex, NULL), &lino, _err6) ; + TAOS_CHECK_GOTO(taosThreadMutexInit(&pTsdb->rCache.writeBatchMutex, NULL), &lino, _err6); pTsdb->rCache.writebatch = writebatch; pTsdb->rCache.my_comparator = cmp; @@ -230,6 +230,9 @@ static int32_t tsdbOpenRocksCache(STsdb *pTsdb) { pTsdb->rCache.readoptions = readoptions; pTsdb->rCache.flushoptions = flushoptions; pTsdb->rCache.db = db; + pTsdb->rCache.sver = -1; + pTsdb->rCache.suid = -1; + pTsdb->rCache.uid = -1; pTsdb->rCache.pTSchema = NULL; TAOS_RETURN(code); @@ -1132,19 +1135,17 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray (void)taosThreadMutexLock(&pTsdb->lruMutex); for (int i = 0; i < num_keys; ++i) { - SLastUpdateCtx *updCtx = (SLastUpdateCtx *)taosArrayGet(updCtxArray, i); - - int8_t lflag = updCtx->lflag; - SRowKey *pRowKey = &updCtx->tsdbRowKey.key; - SColVal *pColVal = &updCtx->colVal; + SLastUpdateCtx *updCtx = &((SLastUpdateCtx *)TARRAY_DATA(updCtxArray))[i]; + int8_t lflag = updCtx->lflag; + SRowKey *pRowKey = &updCtx->tsdbRowKey.key; + SColVal *pColVal = &updCtx->colVal; if (lflag == LFLAG_LAST && !COL_VAL_IS_VALUE(pColVal)) { continue; } SLastKey *key = &(SLastKey){.lflag = lflag, .uid = uid, .cid = pColVal->cid}; - size_t klen = ROCKS_KEY_LEN; - LRUHandle *h = taosLRUCacheLookup(pCache, key, klen); + LRUHandle *h = taosLRUCacheLookup(pCache, key, ROCKS_KEY_LEN); if (h) { SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pCache, h); if (pLastCol->cacheStatus != TSDB_LAST_CACHE_NO_CACHE) { @@ -1299,53 +1300,94 @@ _exit: TAOS_RETURN(code); } +void tsdbCacheInvalidateSchema(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int32_t sver) { + SRocksCache *pRCache = &pTsdb->rCache; + if (!pRCache->pTSchema || sver <= pTsdb->rCache.sver) return; + + if (suid > 0 && suid == pRCache->suid) { + pRCache->sver = -1; + pRCache->suid = -1; + } + if (suid == 0 && uid == pRCache->uid) { + pRCache->sver = -1; + pRCache->uid = -1; + } +} + +static int32_t tsdbUpdateSkm(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int32_t sver) { + SRocksCache *pRCache = &pTsdb->rCache; + if (pRCache->pTSchema && sver == pRCache->sver) { + if (suid > 0 && suid == pRCache->suid) { + return 0; + } + if (suid == 0 && uid == pRCache->uid) { + return 0; + } + } + + pRCache->suid = suid; + pRCache->uid = uid; + pRCache->sver = sver; + tDestroyTSchema(pRCache->pTSchema); + return metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, -1, &pRCache->pTSchema); +} + int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int64_t version, int32_t nRow, SRow **aRow) { int32_t code = 0, lino = 0; // 1. prepare last - TSDBROW lRow = {.type = TSDBROW_ROW_FMT, .pTSRow = aRow[nRow - 1], .version = version}; - + TSDBROW lRow = {.type = TSDBROW_ROW_FMT, .pTSRow = aRow[nRow - 1], .version = version}; STSchema *pTSchema = NULL; int32_t sver = TSDBROW_SVERSION(&lRow); SArray *ctxArray = NULL; SSHashObj *iColHash = NULL; - TAOS_CHECK_GOTO(metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, sver, &pTSchema), &lino, _exit); + TAOS_CHECK_GOTO(tsdbUpdateSkm(pTsdb, suid, uid, sver), &lino, _exit); + pTSchema = pTsdb->rCache.pTSchema; TSDBROW tRow = {.type = TSDBROW_ROW_FMT, .version = version}; int32_t nCol = pTSchema->numOfCols; - ctxArray = taosArrayInit(nCol, sizeof(SLastUpdateCtx)); - iColHash = tSimpleHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT)); + ctxArray = taosArrayInit(nCol * 2, sizeof(SLastUpdateCtx)); + if (ctxArray == NULL) { + TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit); + } // 1. prepare by lrow STsdbRowKey tsdbRowKey = {0}; tsdbRowGetKey(&lRow, &tsdbRowKey); STSDBRowIter iter = {0}; - code = tsdbRowIterOpen(&iter, &lRow, pTSchema); - if (code != TSDB_CODE_SUCCESS) { - tsdbError("vgId:%d, %s tsdbRowIterOpen failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, - tstrerror(code)); - TAOS_CHECK_GOTO(code, &lino, _exit); - } + TAOS_CHECK_GOTO(tsdbRowIterOpen(&iter, &lRow, pTSchema), &lino, _exit); + int32_t iCol = 0; for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal && iCol < nCol; pColVal = tsdbRowIterNext(&iter), iCol++) { SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal}; if (!taosArrayPush(ctxArray, &updateCtx)) { + tsdbRowClose(&iter); TAOS_CHECK_GOTO(terrno, &lino, _exit); } - if (!COL_VAL_IS_VALUE(pColVal)) { + if (COL_VAL_IS_VALUE(pColVal)) { + updateCtx.lflag = LFLAG_LAST; + if (!taosArrayPush(ctxArray, &updateCtx)) { + tsdbRowClose(&iter); + TAOS_CHECK_GOTO(terrno, &lino, _exit); + } + } else { + if (!iColHash) { + iColHash = tSimpleHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT)); + if (iColHash == NULL) { + tsdbRowClose(&iter); + TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit); + } + } + if (tSimpleHashPut(iColHash, &iCol, sizeof(iCol), NULL, 0)) { + tsdbRowClose(&iter); TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit); } - continue; - } - updateCtx.lflag = LFLAG_LAST; - if (!taosArrayPush(ctxArray, &updateCtx)) { - TAOS_CHECK_GOTO(terrno, &lino, _exit); } } tsdbRowClose(&iter); @@ -1390,7 +1432,10 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6 } _exit: - taosMemoryFreeClear(pTSchema); + if (code) { + tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tstrerror(code)); + } + taosArrayDestroy(ctxArray); tSimpleHashCleanup(iColHash); diff --git a/source/util/src/tlrucache.c b/source/util/src/tlrucache.c index 69832cd46c..43a220af50 100644 --- a/source/util/src/tlrucache.c +++ b/source/util/src/tlrucache.c @@ -14,12 +14,12 @@ */ #define _DEFAULT_SOURCE +#include "tlrucache.h" #include "os.h" #include "taoserror.h" #include "tarray.h" #include "tdef.h" #include "tlog.h" -#include "tlrucache.h" #include "tutil.h" typedef struct SLRUEntry SLRUEntry; @@ -110,7 +110,7 @@ struct SLRUEntryTable { }; static int taosLRUEntryTableInit(SLRUEntryTable *table, int maxUpperHashBits) { - table->lengthBits = 4; + table->lengthBits = 16; table->list = taosMemoryCalloc(1 << table->lengthBits, sizeof(SLRUEntry *)); if (!table->list) { TAOS_RETURN(terrno); @@ -371,24 +371,35 @@ static void taosLRUCacheShardCleanup(SLRUCacheShard *shard) { static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *e, LRUHandle **handle, bool freeOnFail) { - LRUStatus status = TAOS_LRU_STATUS_OK; - SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES); - if (!lastReferenceList) { - taosLRUEntryFree(e); - return TAOS_LRU_STATUS_FAIL; + LRUStatus status = TAOS_LRU_STATUS_OK; + SLRUEntry *toFree = NULL; + SArray *lastReferenceList = NULL; + if (shard->usage + e->totalCharge > shard->capacity) { + lastReferenceList = taosArrayInit(16, POINTER_BYTES); + if (!lastReferenceList) { + taosLRUEntryFree(e); + return TAOS_LRU_STATUS_FAIL; + } } (void)taosThreadMutexLock(&shard->mutex); - taosLRUCacheShardEvictLRU(shard, e->totalCharge, lastReferenceList); + if (shard->usage + e->totalCharge > shard->capacity && shard->lru.next != &shard->lru) { + if (!lastReferenceList) { + lastReferenceList = taosArrayInit(16, POINTER_BYTES); + if (!lastReferenceList) { + taosLRUEntryFree(e); + (void)taosThreadMutexUnlock(&shard->mutex); + return TAOS_LRU_STATUS_FAIL; + } + } + taosLRUCacheShardEvictLRU(shard, e->totalCharge, lastReferenceList); + } if (shard->usage + e->totalCharge > shard->capacity && (shard->strictCapacity || handle == NULL)) { TAOS_LRU_ENTRY_SET_IN_CACHE(e, false); if (handle == NULL) { - if (!taosArrayPush(lastReferenceList, &e)) { - taosLRUEntryFree(e); - goto _exit; - } + toFree = e; } else { if (freeOnFail) { taosLRUEntryFree(e); @@ -413,11 +424,7 @@ static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry * taosLRUCacheShardLRURemove(shard, old); shard->usage -= old->totalCharge; - if (!taosArrayPush(lastReferenceList, &old)) { - taosLRUEntryFree(e); - taosLRUEntryFree(old); - goto _exit; - } + toFree = old; } } if (handle == NULL) { @@ -434,6 +441,10 @@ static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry * _exit: (void)taosThreadMutexUnlock(&shard->mutex); + if (toFree) { + taosLRUEntryFree(toFree); + } + for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) { SLRUEntry *entry = taosArrayGetP(lastReferenceList, i); @@ -733,7 +744,8 @@ void taosLRUCacheCleanup(SLRUCache *cache) { } LRUStatus taosLRUCacheInsert(SLRUCache *cache, const void *key, size_t keyLen, void *value, size_t charge, - _taos_lru_deleter_t deleter, _taos_lru_overwriter_t overwriter, LRUHandle **handle, LRUPriority priority, void *ud) { + _taos_lru_deleter_t deleter, _taos_lru_overwriter_t overwriter, LRUHandle **handle, + LRUPriority priority, void *ud) { uint32_t hash = TAOS_LRU_CACHE_SHARD_HASH32(key, keyLen); uint32_t shardIndex = hash & cache->shardedCache.shardMask;