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

Merge pull request #13656 from taosdata/enh/refactorIdx 

refactor: index module
This commit is contained in:
Yihao Deng 2022-06-10 09:36:27 +08:00 committed by GitHub
parent f974d625b6 f057cd1943
commit a4ad2b3f9a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 148 additions and 148 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
source/libs/index/inc/indexCache.h
View File

@ -62,25 +62,25 @@ typedef struct CacheTerm {
} CacheTerm;
//
IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type);
IndexCache* idxCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type);
void indexCacheForceToMerge(void* cache);
void indexCacheDestroy(void* cache);
void indexCacheBroadcast(void* cache);
void indexCacheWait(void* cache);
void idxCacheForceToMerge(void* cache);
void idxCacheDestroy(void* cache);
void idxCacheBroadcast(void* cache);
void idxCacheWait(void* cache);
Iterate* indexCacheIteratorCreate(IndexCache* cache);
Iterate* idxCacheIteratorCreate(IndexCache* cache);
void idxCacheIteratorDestroy(Iterate* iiter);
int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid);
int idxCachePut(void* cache, SIndexTerm* term, uint64_t uid);
// int indexCacheGet(void *cache, uint64_t *rst);
int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* tr, STermValueType* s);
int idxCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* tr, STermValueType* s);
void indexCacheRef(IndexCache* cache);
void indexCacheUnRef(IndexCache* cache);
void idxCacheRef(IndexCache* cache);
void idxCacheUnRef(IndexCache* cache);
void indexCacheDebug(IndexCache* cache);
void idxCacheDebug(IndexCache* cache);
void idxCacheDestroyImm(IndexCache* cache);
#ifdef __cplusplus

4
source/libs/index/inc/indexComm.h
View File

@ -34,11 +34,11 @@ typedef enum { MATCH, CONTINUE, BREAK } TExeCond;
typedef TExeCond (*_cache_range_compare)(void* a, void* b, int8_t type);
__compar_fn_t indexGetCompar(int8_t type);
__compar_fn_t idxGetCompar(int8_t type);
TExeCond tCompare(__compar_fn_t func, int8_t cmpType, void* a, void* b, int8_t dType);
TExeCond tDoCompare(__compar_fn_t func, int8_t cmpType, void* a, void* b);
_cache_range_compare indexGetCompare(RangeType ty);
_cache_range_compare idxGetCompare(RangeType ty);
int32_t idxConvertData(void* src, int8_t type, void** dst);
int32_t idxConvertDataToStr(void* src, int8_t type, void** dst);

24
source/libs/index/inc/indexInt.h
View File

@ -133,24 +133,24 @@ typedef struct TFileCacheKey {
} ICacheKey;
int idxFlushCacheToTFile(SIndex* sIdx, void*, bool quit);
int64_t indexAddRef(void* p);
int32_t indexRemoveRef(int64_t ref);
void indexAcquireRef(int64_t ref);
void indexReleaseRef(int64_t ref);
int64_t idxAddRef(void* p);
int32_t idxRemoveRef(int64_t ref);
void idxAcquireRef(int64_t ref);
void idxReleaseRef(int64_t ref);
int32_t indexSerialCacheKey(ICacheKey* key, char* buf);
int32_t idxSerialCacheKey(ICacheKey* key, char* buf);
// int32_t indexSerialKey(ICacheKey* key, char* buf);
// int32_t indexSerialTermKey(SIndexTerm* itm, char* buf);
#define INDEX_TYPE_CONTAIN_EXTERN_TYPE(ty, exTy) (((ty >> 4) & (exTy)) != 0)
#define IDX_TYPE_CONTAIN_EXTERN_TYPE(ty, exTy) (((ty >> 4) & (exTy)) != 0)
#define INDEX_TYPE_GET_TYPE(ty) (ty & 0x0F)
#define IDX_TYPE_GET_TYPE(ty) (ty & 0x0F)
#define INDEX_TYPE_ADD_EXTERN_TYPE(ty, exTy) \
do { \
uint8_t oldTy = ty; \
ty = (ty >> 4) | exTy; \
ty = (ty << 4) | oldTy; \
#define IDX_TYPE_ADD_EXTERN_TYPE(ty, exTy) \
do { \
uint8_t oldTy = ty; \
ty = (ty >> 4) | exTy; \
ty = (ty << 4) | oldTy; \
} while (0)
#ifdef __cplusplus

8
source/libs/index/inc/indexTfile.h
View File

@ -117,10 +117,10 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order);
int tfileWriterFinish(TFileWriter* tw);
//
IndexTFile* indexTFileCreate(const char* path);
void indexTFileDestroy(IndexTFile* tfile);
int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid);
int indexTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* tr);
IndexTFile* idxTFileCreate(const char* path);
void idxTFileDestroy(IndexTFile* tfile);
int idxTFilePut(void* tfile, SIndexTerm* term, uint64_t uid);
int idxTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* tr);
Iterate* tfileIteratorCreate(TFileReader* reader);
void tfileIteratorDestroy(Iterate* iterator);

68
source/libs/index/src/index.c
View File

@ -90,7 +90,7 @@ static void idxMergeCacheAndTFile(SArray* result, IterateValue* icache, IterateV
// static int32_t indexSerialTermKey(SIndexTerm* itm, char* buf);
// int32_t indexSerialKey(ICacheKey* key, char* buf);
static void indexPost(void* idx) {
static void idxPost(void* idx) {
SIndex* pIdx = idx;
tsem_post(&pIdx->sem);
}
@ -106,8 +106,8 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
return -1;
}
// sIdx->cache = (void*)indexCacheCreate(sIdx);
sIdx->tindex = indexTFileCreate(path);
// sIdx->cache = (void*)idxCacheCreate(sIdx);
sIdx->tindex = idxTFileCreate(path);
if (sIdx->tindex == NULL) {
goto END;
}
@ -118,8 +118,8 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
taosThreadMutexInit(&sIdx->mtx, NULL);
tsem_init(&sIdx->sem, 0, 0);
sIdx->refId = indexAddRef(sIdx);
indexAcquireRef(sIdx->refId);
sIdx->refId = idxAddRef(sIdx);
idxAcquireRef(sIdx->refId);
*index = sIdx;
return 0;
@ -136,7 +136,7 @@ void indexDestroy(void* handle) {
SIndex* sIdx = handle;
taosThreadMutexDestroy(&sIdx->mtx);
tsem_destroy(&sIdx->sem);
indexTFileDestroy(sIdx->tindex);
idxTFileDestroy(sIdx->tindex);
taosMemoryFree(sIdx->path);
taosMemoryFree(sIdx);
return;
@ -147,33 +147,33 @@ void indexClose(SIndex* sIdx) {
void* iter = taosHashIterate(sIdx->colObj, NULL);
while (iter) {
IndexCache** pCache = iter;
indexCacheForceToMerge((void*)(*pCache));
idxCacheForceToMerge((void*)(*pCache));
indexInfo("%s wait to merge", (*pCache)->colName);
indexWait((void*)(sIdx));
indexInfo("%s finish to wait", (*pCache)->colName);
iter = taosHashIterate(sIdx->colObj, iter);
indexCacheUnRef(*pCache);
idxCacheUnRef(*pCache);
}
taosHashCleanup(sIdx->colObj);
sIdx->colObj = NULL;
}
indexReleaseRef(sIdx->refId);
indexRemoveRef(sIdx->refId);
idxReleaseRef(sIdx->refId);
idxRemoveRef(sIdx->refId);
}
int64_t indexAddRef(void* p) {
int64_t idxAddRef(void* p) {
// impl
return taosAddRef(indexRefMgt, p);
}
int32_t indexRemoveRef(int64_t ref) {
int32_t idxRemoveRef(int64_t ref) {
// impl later
return taosRemoveRef(indexRefMgt, ref);
}
void indexAcquireRef(int64_t ref) {
void idxAcquireRef(int64_t ref) {
// impl
taosAcquireRef(indexRefMgt, ref);
}
void indexReleaseRef(int64_t ref) {
void idxReleaseRef(int64_t ref) {
// impl
taosReleaseRef(indexRefMgt, ref);
}
@ -186,11 +186,11 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
char buf[128] = {0};
ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName), .colType = p->colType};
int32_t sz = indexSerialCacheKey(&key, buf);
int32_t sz = idxSerialCacheKey(&key, buf);
IndexCache** cache = taosHashGet(index->colObj, buf, sz);
if (cache == NULL) {
IndexCache* pCache = indexCacheCreate(index, p->suid, p->colName, p->colType);
IndexCache* pCache = idxCacheCreate(index, p->suid, p->colName, p->colType);
taosHashPut(index->colObj, buf, sz, &pCache, sizeof(void*));
}
}
@ -201,12 +201,12 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
char buf[128] = {0};
ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName), .colType = p->colType};
int32_t sz = indexSerialCacheKey(&key, buf);
int32_t sz = idxSerialCacheKey(&key, buf);
indexDebug("w suid: %" PRIu64 ", colName: %s, colType: %d", key.suid, key.colName, key.colType);
IndexCache** cache = taosHashGet(index->colObj, buf, sz);
assert(*cache != NULL);
int ret = indexCachePut(*cache, p, uid);
int ret = idxCachePut(*cache, p, uid);
if (ret != 0) {
return ret;
}
@ -289,7 +289,7 @@ SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn oper, uint8_t colTy
tm->nColName = nColName;
char* buf = NULL;
int32_t len = idxConvertDataToStr((void*)colVal, INDEX_TYPE_GET_TYPE(colType), (void**)&buf);
int32_t len = idxConvertDataToStr((void*)colVal, IDX_TYPE_GET_TYPE(colType), (void**)&buf);
assert(len != -1);
tm->colVal = buf;
@ -331,7 +331,7 @@ static int idxTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result)
ICacheKey key = {
.suid = term->suid, .colName = term->colName, .nColName = strlen(term->colName), .colType = term->colType};
indexDebug("r suid: %" PRIu64 ", colName: %s, colType: %d", key.suid, key.colName, key.colType);
int32_t sz = indexSerialCacheKey(&key, buf);
int32_t sz = idxSerialCacheKey(&key, buf);
taosThreadMutexLock(&sIdx->mtx);
IndexCache** pCache = taosHashGet(sIdx->colObj, buf, sz);
@ -345,14 +345,14 @@ static int idxTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result)
int64_t st = taosGetTimestampUs();
SIdxTRslt* tr = idxTRsltCreate();
if (0 == indexCacheSearch(cache, query, tr, &s)) {
if (0 == idxCacheSearch(cache, query, tr, &s)) {
if (s == kTypeDeletion) {
indexInfo("col: %s already drop by", term->colName);
// coloum already drop by other oper, no need to query tindex
return 0;
} else {
st = taosGetTimestampUs();
if (0 != indexTFileSearch(sIdx->tindex, query, tr)) {
if (0 != idxTFileSearch(sIdx->tindex, query, tr)) {
indexError("corrupt at index(TFile) col:%s val: %s", term->colName, term->colVal);
goto END;
}
@ -465,23 +465,23 @@ int idxFlushCacheToTFile(SIndex* sIdx, void* cache, bool quit) {
IndexCache* pCache = (IndexCache*)cache;
while (quit && atomic_load_32(&pCache->merging) == 1) {
}
while (quit && atomic_load_32(&pCache->merging) == 1)
;
TFileReader* pReader = tfileGetReaderByCol(sIdx->tindex, pCache->suid, pCache->colName);
if (pReader == NULL) {
indexWarn("empty tfile reader found");
}
// handle flush
Iterate* cacheIter = indexCacheIteratorCreate(pCache);
Iterate* cacheIter = idxCacheIteratorCreate(pCache);
if (cacheIter == NULL) {
indexError("%p immtable is empty, ignore merge opera", pCache);
idxCacheDestroyImm(pCache);
tfileReaderUnRef(pReader);
atomic_store_32(&pCache->merging, 0);
if (quit) {
indexPost(sIdx);
idxPost(sIdx);
}
indexReleaseRef(sIdx->refId);
idxReleaseRef(sIdx->refId);
return 0;
}
@ -532,7 +532,7 @@ int idxFlushCacheToTFile(SIndex* sIdx, void* cache, bool quit) {
tfileIteratorDestroy(tfileIter);
tfileReaderUnRef(pReader);
indexCacheUnRef(pCache);
idxCacheUnRef(pCache);
int64_t cost = taosGetTimestampUs() - st;
if (ret != 0) {
@ -542,9 +542,9 @@ int idxFlushCacheToTFile(SIndex* sIdx, void* cache, bool quit) {
}
atomic_store_32(&pCache->merging, 0);
if (quit) {
indexPost(sIdx);
idxPost(sIdx);
}
indexReleaseRef(sIdx->refId);
idxReleaseRef(sIdx->refId);
return ret;
}
@ -561,7 +561,7 @@ void iterateValueDestroy(IterateValue* value, bool destroy) {
value->colVal = NULL;
}
static int64_t indexGetAvaialbleVer(SIndex* sIdx, IndexCache* cache) {
static int64_t idxGetAvailableVer(SIndex* sIdx, IndexCache* cache) {
ICacheKey key = {.suid = cache->suid, .colName = cache->colName, .nColName = strlen(cache->colName)};
int64_t ver = CACHE_VERSION(cache);
@ -579,7 +579,7 @@ static int64_t indexGetAvaialbleVer(SIndex* sIdx, IndexCache* cache) {
return ver;
}
static int idxGenTFile(SIndex* sIdx, IndexCache* cache, SArray* batch) {
int64_t version = indexGetAvaialbleVer(sIdx, cache);
int64_t version = idxGetAvailableVer(sIdx, cache);
indexInfo("file name version: %" PRId64 "", version);
uint8_t colType = cache->type;
@ -620,8 +620,8 @@ END:
return -1;
}
int32_t indexSerialCacheKey(ICacheKey* key, char* buf) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(key->colType, TSDB_DATA_TYPE_JSON);
int32_t idxSerialCacheKey(ICacheKey* key, char* buf) {
bool hasJson = IDX_TYPE_CONTAIN_EXTERN_TYPE(key->colType, TSDB_DATA_TYPE_JSON);
char* p = buf;
char tbuf[65] = {0};

104
source/libs/index/src/indexCache.c
View File

@ -68,7 +68,7 @@ static int32_t (*cacheSearch[][QUERY_MAX])(void* cache, SIndexTerm* ct, SIdxTRsl
cacheSearchLessThan_JSON, cacheSearchLessEqual_JSON, cacheSearchGreaterThan_JSON, cacheSearchGreaterEqual_JSON,
cacheSearchRange_JSON}};
static void doMergeWork(SSchedMsg* msg);
static void idxDoMergeWork(SSchedMsg* msg);
static bool idxCacheIteratorNext(Iterate* itera);
static int32_t cacheSearchTerm(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
@ -127,7 +127,7 @@ static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* term, SIdxTRslt*
MemTable* mem = cache;
IndexCache* pCache = mem->pCache;
_cache_range_compare cmpFn = indexGetCompare(type);
_cache_range_compare cmpFn = idxGetCompare(type);
CacheTerm* pCt = taosMemoryCalloc(1, sizeof(CacheTerm));
pCt->colVal = term->colVal;
@ -187,7 +187,7 @@ static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr
pCt->version = atomic_load_64(&pCache->version);
char* exBuf = NULL;
if (INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON)) {
if (IDX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON)) {
exBuf = idxPackJsonData(term);
pCt->colVal = exBuf;
}
@ -257,7 +257,7 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTR
if (cache == NULL) {
return 0;
}
_cache_range_compare cmpFn = indexGetCompare(type);
_cache_range_compare cmpFn = idxGetCompare(type);
MemTable* mem = cache;
IndexCache* pCache = mem->pCache;
@ -266,7 +266,7 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTR
pCt->colVal = term->colVal;
pCt->version = atomic_load_64(&pCache->version);
int8_t dType = INDEX_TYPE_GET_TYPE(term->colType);
int8_t dType = IDX_TYPE_GET_TYPE(term->colType);
int skip = 0;
char* exBuf = NULL;
if (type == CONTAINS) {
@ -331,9 +331,9 @@ static int32_t cacheSearchRange(void* cache, SIndexTerm* term, SIdxTRslt* tr, ST
// impl later
return 0;
}
static IterateValue* indexCacheIteratorGetValue(Iterate* iter);
static IterateValue* idxCacheIteratorGetValue(Iterate* iter);
IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type) {
IndexCache* idxCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type) {
IndexCache* cache = taosMemoryCalloc(1, sizeof(IndexCache));
if (cache == NULL) {
indexError("failed to create index cache");
@ -342,7 +342,7 @@ IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, in
cache->mem = idxInternalCacheCreate(type);
cache->mem->pCache = cache;
cache->colName = INDEX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? tstrdup(JSON_COLUMN) : tstrdup(colName);
cache->colName = IDX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? tstrdup(JSON_COLUMN) : tstrdup(colName);
cache->type = type;
cache->index = idx;
cache->version = 0;
@ -352,13 +352,13 @@ IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, in
taosThreadMutexInit(&cache->mtx, NULL);
taosThreadCondInit(&cache->finished, NULL);
indexCacheRef(cache);
idxCacheRef(cache);
if (idx != NULL) {
indexAcquireRef(idx->refId);
idxAcquireRef(idx->refId);
}
return cache;
}
void indexCacheDebug(IndexCache* cache) {
void idxCacheDebug(IndexCache* cache) {
MemTable* tbl = NULL;
taosThreadMutexLock(&cache->mtx);
@ -405,7 +405,7 @@ void indexCacheDebug(IndexCache* cache) {
}
}
void indexCacheDestroySkiplist(SSkipList* slt) {
void idxCacheDestroySkiplist(SSkipList* slt) {
SSkipListIterator* iter = tSkipListCreateIter(slt);
while (iter != NULL && tSkipListIterNext(iter)) {
SSkipListNode* node = tSkipListIterGet(iter);
@ -418,11 +418,11 @@ void indexCacheDestroySkiplist(SSkipList* slt) {
tSkipListDestroyIter(iter);
tSkipListDestroy(slt);
}
void indexCacheBroadcast(void* cache) {
void idxCacheBroadcast(void* cache) {
IndexCache* pCache = cache;
taosThreadCondBroadcast(&pCache->finished);
}
void indexCacheWait(void* cache) {
void idxCacheWait(void* cache) {
IndexCache* pCache = cache;
taosThreadCondWait(&pCache->finished, &pCache->mtx);
}
@ -435,14 +435,14 @@ void idxCacheDestroyImm(IndexCache* cache) {
tbl = cache->imm;
cache->imm = NULL; // or throw int bg thread
indexCacheBroadcast(cache);
idxCacheBroadcast(cache);
taosThreadMutexUnlock(&cache->mtx);
idxMemUnRef(tbl);
idxMemUnRef(tbl);
}
void indexCacheDestroy(void* cache) {
void idxCacheDestroy(void* cache) {
IndexCache* pCache = cache;
if (pCache == NULL) {
return;
@ -455,12 +455,12 @@ void indexCacheDestroy(void* cache) {
taosThreadMutexDestroy(&pCache->mtx);
taosThreadCondDestroy(&pCache->finished);
if (pCache->index != NULL) {
indexReleaseRef(((SIndex*)pCache->index)->refId);
idxReleaseRef(((SIndex*)pCache->index)->refId);
}
taosMemoryFree(pCache);
}
Iterate* indexCacheIteratorCreate(IndexCache* cache) {
Iterate* idxCacheIteratorCreate(IndexCache* cache) {
if (cache->imm == NULL) {
return NULL;
}
@ -477,7 +477,7 @@ Iterate* indexCacheIteratorCreate(IndexCache* cache) {
iiter->val.colVal = NULL;
iiter->iter = tbl != NULL ? tSkipListCreateIter(tbl->mem) : NULL;
iiter->next = idxCacheIteratorNext;
iiter->getValue = indexCacheIteratorGetValue;
iiter->getValue = idxCacheIteratorGetValue;
taosThreadMutexUnlock(&cache->mtx);
@ -492,30 +492,30 @@ void idxCacheIteratorDestroy(Iterate* iter) {
taosMemoryFree(iter);
}
int indexCacheSchedToMerge(IndexCache* pCache, bool notify) {
int idxCacheSchedToMerge(IndexCache* pCache, bool notify) {
SSchedMsg schedMsg = {0};
schedMsg.fp = doMergeWork;
schedMsg.fp = idxDoMergeWork;
schedMsg.ahandle = pCache;
if (notify) {
schedMsg.thandle = taosMemoryMalloc(1);
}
schedMsg.msg = NULL;
indexAcquireRef(pCache->index->refId);
idxAcquireRef(pCache->index->refId);
taosScheduleTask(indexQhandle, &schedMsg);
return 0;
}
static void indexCacheMakeRoomForWrite(IndexCache* cache) {
static void idxCacheMakeRoomForWrite(IndexCache* cache) {
while (true) {
if (cache->occupiedMem * MEM_ESTIMATE_RADIO < MEM_THRESHOLD) {
break;
} else if (cache->imm != NULL) {
// TODO: wake up by condition variable
indexCacheWait(cache);
idxCacheWait(cache);
} else {
bool quit = cache->occupiedMem >= MEM_SIGNAL_QUIT ? true : false;
indexCacheRef(cache);
idxCacheRef(cache);
cache->imm = cache->mem;
cache->mem = idxInternalCacheCreate(cache->type);
cache->mem->pCache = cache;
@ -525,18 +525,18 @@ static void indexCacheMakeRoomForWrite(IndexCache* cache) {
}
// sched to merge
// unref cache in bgwork
indexCacheSchedToMerge(cache, quit);
idxCacheSchedToMerge(cache, quit);
}
}
}
int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid) {
int idxCachePut(void* cache, SIndexTerm* term, uint64_t uid) {
if (cache == NULL) {
return -1;
}
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON);
bool hasJson = IDX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON);
IndexCache* pCache = cache;
indexCacheRef(pCache);
idxCacheRef(pCache);
// encode data
CacheTerm* ct = taosMemoryCalloc(1, sizeof(CacheTerm));
if (cache == NULL) {
@ -559,7 +559,7 @@ int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid) {
taosThreadMutexLock(&pCache->mtx);
pCache->occupiedMem += estimate;
indexCacheMakeRoomForWrite(pCache);
idxCacheMakeRoomForWrite(pCache);
MemTable* tbl = pCache->mem;
idxMemRef(tbl);
tSkipListPut(tbl->mem, (char*)ct);
@ -567,29 +567,29 @@ int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid) {
taosThreadMutexUnlock(&pCache->mtx);
indexCacheUnRef(pCache);
idxCacheUnRef(pCache);
return 0;
// encode end
}
void indexCacheForceToMerge(void* cache) {
void idxCacheForceToMerge(void* cache) {
IndexCache* pCache = cache;
indexCacheRef(pCache);
idxCacheRef(pCache);
taosThreadMutexLock(&pCache->mtx);
indexInfo("%p is forced to merge into tfile", pCache);
pCache->occupiedMem += MEM_SIGNAL_QUIT;
indexCacheMakeRoomForWrite(pCache);
idxCacheMakeRoomForWrite(pCache);
taosThreadMutexUnlock(&pCache->mtx);
indexCacheUnRef(pCache);
idxCacheUnRef(pCache);
return;
}
int indexCacheDel(void* cache, const char* fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) {
int idxCacheDel(void* cache, const char* fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) {
IndexCache* pCache = cache;
return 0;
}
static int32_t indexQueryMem(MemTable* mem, SIndexTermQuery* query, SIdxTRslt* tr, STermValueType* s) {
static int32_t idxQueryMem(MemTable* mem, SIndexTermQuery* query, SIdxTRslt* tr, STermValueType* s) {
if (mem == NULL) {
return 0;
}
@ -597,13 +597,13 @@ static int32_t indexQueryMem(MemTable* mem, SIndexTermQuery* query, SIdxTRslt* t
SIndexTerm* term = query->term;
EIndexQueryType qtype = query->qType;
if (INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON)) {
if (IDX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON)) {
return cacheSearch[1][qtype](mem, term, tr, s);
} else {
return cacheSearch[0][qtype](mem, term, tr, s);
}
}
int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* result, STermValueType* s) {
int idxCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* result, STermValueType* s) {
int64_t st = taosGetTimestampUs();
if (cache == NULL) {
return 0;
@ -618,10 +618,10 @@ int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* result, STe
idxMemRef(imm);
taosThreadMutexUnlock(&pCache->mtx);
int ret = (mem && mem->mem) ? indexQueryMem(mem, query, result, s) : 0;
int ret = (mem && mem->mem) ? idxQueryMem(mem, query, result, s) : 0;
if (ret == 0 && *s != kTypeDeletion) {
// continue search in imm
ret = (imm && imm->mem) ? indexQueryMem(imm, query, result, s) : 0;
ret = (imm && imm->mem) ? idxQueryMem(imm, query, result, s) : 0;
}
idxMemUnRef(mem);
@ -631,20 +631,20 @@ int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* result, STe
return ret;
}
void indexCacheRef(IndexCache* cache) {
void idxCacheRef(IndexCache* cache) {
if (cache == NULL) {
return;
}
int ref = T_REF_INC(cache);
UNUSED(ref);
}
void indexCacheUnRef(IndexCache* cache) {
void idxCacheUnRef(IndexCache* cache) {
if (cache == NULL) {
return;
}
int ref = T_REF_DEC(cache);
if (ref == 0) {
indexCacheDestroy(cache);
idxCacheDestroy(cache);
}
}
@ -662,7 +662,7 @@ void idxMemUnRef(MemTable* tbl) {
int ref = T_REF_DEC(tbl);
if (ref == 0) {
SSkipList* slt = tbl->mem;
indexCacheDestroySkiplist(slt);
idxCacheDestroySkiplist(slt);
taosMemoryFree(tbl);
}
}
@ -693,15 +693,15 @@ static int32_t idxCacheTermCompare(const void* l, const void* r) {
return cmp;
}
static int indexFindCh(char* a, char c) {
static int idxFindCh(char* a, char c) {
char* p = a;
while (*p != 0 && *p++ != c) {
}
return p - a;
}
static int idxCacheJsonTermCompareImpl(char* a, char* b) {
// int alen = indexFindCh(a, '&');
// int blen = indexFindCh(b, '&');
// int alen = idxFindCh(a, '&');
// int blen = idxFindCh(b, '&');
// int cmp = strncmp(a, b, MIN(alen, blen));
// if (cmp == 0) {
@ -730,9 +730,9 @@ static int32_t idxCacheJsonTermCompare(const void* l, const void* r) {
return cmp;
}
static MemTable* idxInternalCacheCreate(int8_t type) {
int ttype = INDEX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? TSDB_DATA_TYPE_BINARY : TSDB_DATA_TYPE_BINARY;
int ttype = IDX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? TSDB_DATA_TYPE_BINARY : TSDB_DATA_TYPE_BINARY;
int32_t (*cmpFn)(const void* l, const void* r) =
INDEX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? idxCacheJsonTermCompare : idxCacheTermCompare;
IDX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? idxCacheJsonTermCompare : idxCacheTermCompare;
MemTable* tbl = taosMemoryCalloc(1, sizeof(MemTable));
idxMemRef(tbl);
@ -742,7 +742,7 @@ static MemTable* idxInternalCacheCreate(int8_t type) {
return tbl;
}
static void doMergeWork(SSchedMsg* msg) {
static void idxDoMergeWork(SSchedMsg* msg) {
IndexCache* pCache = msg->ahandle;
SIndex* sidx = (SIndex*)pCache->index;
@ -771,7 +771,7 @@ static bool idxCacheIteratorNext(Iterate* itera) {
return next;
}
static IterateValue* indexCacheIteratorGetValue(Iterate* iter) {
static IterateValue* idxCacheIteratorGetValue(Iterate* iter) {
// opt later
return &iter->val;
}

22
source/libs/index/src/indexComm.c
View File

@ -75,35 +75,35 @@ char* idxInt2str(int64_t val, char* dst, int radix) {
;
return dst - 1;
}
__compar_fn_t indexGetCompar(int8_t type) {
__compar_fn_t idxGetCompar(int8_t type) {
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
return (__compar_fn_t)strcmp;
}
return getComparFunc(type, 0);
}
static TExeCond tCompareLessThan(void* a, void* b, int8_t type) {
__compar_fn_t func = indexGetCompar(type);
__compar_fn_t func = idxGetCompar(type);
return tCompare(func, QUERY_LESS_THAN, a, b, type);
}
static TExeCond tCompareLessEqual(void* a, void* b, int8_t type) {
__compar_fn_t func = indexGetCompar(type);
__compar_fn_t func = idxGetCompar(type);
return tCompare(func, QUERY_LESS_EQUAL, a, b, type);
}
static TExeCond tCompareGreaterThan(void* a, void* b, int8_t type) {
__compar_fn_t func = indexGetCompar(type);
__compar_fn_t func = idxGetCompar(type);
return tCompare(func, QUERY_GREATER_THAN, a, b, type);
}
static TExeCond tCompareGreaterEqual(void* a, void* b, int8_t type) {
__compar_fn_t func = indexGetCompar(type);
__compar_fn_t func = idxGetCompar(type);
return tCompare(func, QUERY_GREATER_EQUAL, a, b, type);
}
static TExeCond tCompareContains(void* a, void* b, int8_t type) {
__compar_fn_t func = indexGetCompar(type);
__compar_fn_t func = idxGetCompar(type);
return tCompare(func, QUERY_TERM, a, b, type);
}
static TExeCond tCompareEqual(void* a, void* b, int8_t type) {
__compar_fn_t func = indexGetCompar(type);
__compar_fn_t func = idxGetCompar(type);
return tCompare(func, QUERY_TERM, a, b, type);
}
TExeCond tCompare(__compar_fn_t func, int8_t cmptype, void* a, void* b, int8_t dtype) {
@ -205,14 +205,14 @@ TExeCond tDoCompare(__compar_fn_t func, int8_t comparType, void* a, void* b) {
static TExeCond (*rangeCompare[])(void* a, void* b, int8_t type) = {
tCompareLessThan, tCompareLessEqual, tCompareGreaterThan, tCompareGreaterEqual, tCompareContains, tCompareEqual};
_cache_range_compare indexGetCompare(RangeType ty) { return rangeCompare[ty]; }
_cache_range_compare idxGetCompare(RangeType ty) { return rangeCompare[ty]; }
char* idxPackJsonData(SIndexTerm* itm) {
/*
* |<-----colname---->|<-----dataType---->|<--------colVal---------->|
* |<-----string----->|<-----uint8_t----->|<----depend on dataType-->|
*/
uint8_t ty = INDEX_TYPE_GET_TYPE(itm->colType);
uint8_t ty = IDX_TYPE_GET_TYPE(itm->colType);
int32_t sz = itm->nColName + itm->nColVal + sizeof(uint8_t) + sizeof(JSON_VALUE_DELIM) * 2 + 1;
char* buf = (char*)taosMemoryCalloc(1, sz);
@ -240,7 +240,7 @@ char* idxPackJsonDataPrefix(SIndexTerm* itm, int32_t* skip) {
* |<-----colname---->|<-----dataType---->|<--------colVal---------->|
* |<-----string----->|<-----uint8_t----->|<----depend on dataType-->|
*/
uint8_t ty = INDEX_TYPE_GET_TYPE(itm->colType);
uint8_t ty = IDX_TYPE_GET_TYPE(itm->colType);
int32_t sz = itm->nColName + itm->nColVal + sizeof(uint8_t) + sizeof(JSON_VALUE_DELIM) * 2 + 1;
char* buf = (char*)taosMemoryCalloc(1, sz);
@ -267,7 +267,7 @@ char* idxPackJsonDataPrefixNoType(SIndexTerm* itm, int32_t* skip) {
* |<-----colname---->|<-----dataType---->|<--------colVal---------->|
* |<-----string----->|<-----uint8_t----->|<----depend on dataType-->|
*/
uint8_t ty = INDEX_TYPE_GET_TYPE(itm->colType);
uint8_t ty = IDX_TYPE_GET_TYPE(itm->colType);
int32_t sz = itm->nColName + itm->nColVal + sizeof(uint8_t) + sizeof(JSON_VALUE_DELIM) * 2 + 1;
char* buf = (char*)taosMemoryCalloc(1, sz);

2
source/libs/index/src/indexFilter.c
View File

@ -318,7 +318,7 @@ int sifLessThan(void *a, void *b, int16_t dtype) {
}
int sifEqual(void *a, void *b, int16_t dtype) {
__compar_fn_t func = getComparFunc(dtype, 0);
//__compar_fn_t func = indexGetCompar(dtype);
//__compar_fn_t func = idxGetCompar(dtype);
return (int)tDoCompare(func, QUERY_TERM, a, b);
}
static Filter sifGetFilterFunc(EIndexQueryType type, bool *reverse) {

4
source/libs/index/src/indexJson.c
View File

@ -30,7 +30,7 @@ int indexJsonPut(SIndexJson *index, SIndexJsonMultiTerm *terms, uint64_t uid) {
} else {
p->colType = TSDB_DATA_TYPE_DOUBLE;
}
INDEX_TYPE_ADD_EXTERN_TYPE(p->colType, TSDB_DATA_TYPE_JSON);
IDX_TYPE_ADD_EXTERN_TYPE(p->colType, TSDB_DATA_TYPE_JSON);
}
// handle put
return indexPut(index, terms, uid);
@ -48,7 +48,7 @@ int indexJsonSearch(SIndexJson *index, SIndexJsonMultiTermQuery *tq, SArray *res
} else {
p->colType = TSDB_DATA_TYPE_DOUBLE;
}
INDEX_TYPE_ADD_EXTERN_TYPE(p->colType, TSDB_DATA_TYPE_JSON);
IDX_TYPE_ADD_EXTERN_TYPE(p->colType, TSDB_DATA_TYPE_JSON);
}
// handle search
return indexSearch(index, tq, result);

28
source/libs/index/src/indexTfile.c
View File

@ -118,7 +118,7 @@ TFileCache* tfileCacheCreate(const char* path) {
ICacheKey key = {.suid = header->suid, .colName = header->colName, .nColName = (int32_t)strlen(header->colName)};
char buf[128] = {0};
int32_t sz = indexSerialCacheKey(&key, buf);
int32_t sz = idxSerialCacheKey(&key, buf);
assert(sz < sizeof(buf));
taosHashPut(tcache->tableCache, buf, sz, &reader, sizeof(void*));
tfileReaderRef(reader);
@ -149,7 +149,7 @@ void tfileCacheDestroy(TFileCache* tcache) {
TFileReader* tfileCacheGet(TFileCache* tcache, ICacheKey* key) {
char buf[128] = {0};
int32_t sz = indexSerialCacheKey(key, buf);
int32_t sz = idxSerialCacheKey(key, buf);
assert(sz < sizeof(buf));
TFileReader** reader = taosHashGet(tcache->tableCache, buf, sz);
if (reader == NULL || *reader == NULL) {
@ -161,7 +161,7 @@ TFileReader* tfileCacheGet(TFileCache* tcache, ICacheKey* key) {
}
void tfileCachePut(TFileCache* tcache, ICacheKey* key, TFileReader* reader) {
char buf[128] = {0};
int32_t sz = indexSerialCacheKey(key, buf);
int32_t sz = idxSerialCacheKey(key, buf);
// remove last version index reader
TFileReader** p = taosHashGet(tcache->tableCache, buf, sz);
if (p != NULL && *p != NULL) {
@ -281,7 +281,7 @@ static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return 0;
}
static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
bool hasJson = IDX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
int ret = 0;
char* p = tem->colVal;
@ -305,7 +305,7 @@ static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTRslt* tr,
int ret = 0;
char* p = tem->colVal;
int skip = 0;
_cache_range_compare cmpFn = indexGetCompare(type);
_cache_range_compare cmpFn = idxGetCompare(type);
SArray* offsets = taosArrayInit(16, sizeof(uint64_t));
@ -431,7 +431,7 @@ static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTRslt
p = idxPackJsonDataPrefix(tem, &skip);
}
_cache_range_compare cmpFn = indexGetCompare(ctype);
_cache_range_compare cmpFn = idxGetCompare(ctype);
SArray* offsets = taosArrayInit(16, sizeof(uint64_t));
@ -457,7 +457,7 @@ static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTRslt
} else if (0 != strncmp(ch, p, skip)) {
continue;
}
cond = cmpFn(ch + skip, tem->colVal, INDEX_TYPE_GET_TYPE(tem->colType));
cond = cmpFn(ch + skip, tem->colVal, IDX_TYPE_GET_TYPE(tem->colType));
}
if (MATCH == cond) {
tfileReaderLoadTableIds((TFileReader*)reader, rt->out.out, tr->total);
@ -476,7 +476,7 @@ int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTRslt* tr
SIndexTerm* term = query->term;
EIndexQueryType qtype = query->qType;
int ret = 0;
if (INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON)) {
if (IDX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON)) {
ret = tfSearch[1][qtype](reader, term, tr);
} else {
ret = tfSearch[0][qtype](reader, term, tr);
@ -536,7 +536,7 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
__compar_fn_t fn;
int8_t colType = tw->header.colType;
colType = INDEX_TYPE_GET_TYPE(colType);
colType = IDX_TYPE_GET_TYPE(colType);
if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) {
fn = tfileStrCompare;
} else {
@ -620,7 +620,7 @@ void tfileWriterDestroy(TFileWriter* tw) {
taosMemoryFree(tw);
}
IndexTFile* indexTFileCreate(const char* path) {
IndexTFile* idxTFileCreate(const char* path) {
TFileCache* cache = tfileCacheCreate(path);
if (cache == NULL) {
return NULL;
@ -635,7 +635,7 @@ IndexTFile* indexTFileCreate(const char* path) {
tfile->cache = cache;
return tfile;
}
void indexTFileDestroy(IndexTFile* tfile) {
void idxTFileDestroy(IndexTFile* tfile) {
if (tfile == NULL) {
return;
}
@ -644,7 +644,7 @@ void indexTFileDestroy(IndexTFile* tfile) {
taosMemoryFree(tfile);
}
int indexTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* result) {
int idxTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* result) {
int ret = -1;
if (tfile == NULL) {
return ret;
@ -667,7 +667,7 @@ int indexTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* result) {
return tfileReaderSearch(reader, query, result);
}
int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid) {
int idxTFilePut(void* tfile, SIndexTerm* term, uint64_t uid) {
// TFileWriterOpt wOpt = {.suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName =
// term->nColName, .version = 1};
@ -845,7 +845,7 @@ static int tfileWriteData(TFileWriter* write, TFileValue* tval) {
TFileHeader* header = &write->header;
uint8_t colType = header->colType;
colType = INDEX_TYPE_GET_TYPE(colType);
colType = IDX_TYPE_GET_TYPE(colType);
FstSlice key = fstSliceCreate((uint8_t*)(tval->colVal), (size_t)strlen(tval->colVal));
if (fstBuilderInsert(write->fb, key, tval->offset)) {
fstSliceDestroy(&key);

10
source/libs/index/test/indexTests.cc
View File

@ -521,10 +521,10 @@ class CacheObj {
public:
CacheObj() {
// TODO
cache = indexCacheCreate(NULL, 0, "voltage", TSDB_DATA_TYPE_BINARY);
cache = idxCacheCreate(NULL, 0, "voltage", TSDB_DATA_TYPE_BINARY);
}
int Put(SIndexTerm* term, int16_t colId, int32_t version, uint64_t uid) {
int ret = indexCachePut(cache, term, uid);
int ret = idxCachePut(cache, term, uid);
if (ret != 0) {
//
std::cout << "failed to put into cache: " << ret << std::endl;
@ -533,12 +533,12 @@ class CacheObj {
}
void Debug() {
//
indexCacheDebug(cache);
idxCacheDebug(cache);
}
int Get(SIndexTermQuery* query, int16_t colId, int32_t version, SArray* result, STermValueType* s) {
SIdxTRslt* tr = idxTRsltCreate();
int ret = indexCacheSearch(cache, query, tr, s);
int ret = idxCacheSearch(cache, query, tr, s);
idxTRsltMergeTo(tr, result);
idxTRsltDestroy(tr);
@ -549,7 +549,7 @@ class CacheObj {
}
~CacheObj() {
// TODO
indexCacheDestroy(cache);
idxCacheDestroy(cache);
}
private: