diff --git a/source/libs/index/inc/indexInt.h b/source/libs/index/inc/indexInt.h index 048c9e804e6114017f2f91ed7bb2e70c50b52841..6e1256d857ee0d4accbf67912354e94d1e681df5 100644 --- a/source/libs/index/inc/indexInt.h +++ b/source/libs/index/inc/indexInt.h @@ -108,8 +108,17 @@ void iterateValueDestroy(IterateValue* iv, bool destroy); extern void* indexQhandle; +typedef struct TFileCacheKey { + uint64_t suid; + uint8_t colType; + char* colName; + int32_t nColName; +} ICacheKey; + int indexFlushCacheTFile(SIndex* sIdx, void*); +int32_t indexSerialCacheKey(ICacheKey* key, char* buf); + #define indexFatal(...) \ do { \ if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("index FATAL ", 255, __VA_ARGS__); } \ diff --git a/source/libs/index/inc/index_cache.h b/source/libs/index/inc/index_cache.h index 12b66bca2cab9ff4b658f7aeaacd9268facca6a3..805137ccaf8e95885b77e1d440372d8141bf77e7 100644 --- a/source/libs/index/inc/index_cache.h +++ b/source/libs/index/inc/index_cache.h @@ -42,6 +42,7 @@ typedef struct IndexCache { int32_t version; int32_t nTerm; int8_t type; + uint64_t suid; pthread_mutex_t mtx; } IndexCache; @@ -58,7 +59,7 @@ typedef struct CacheTerm { } CacheTerm; // -IndexCache* indexCacheCreate(SIndex* idx, const char* colName, int8_t type); +IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type); void indexCacheDestroy(void* cache); diff --git a/source/libs/index/inc/index_tfile.h b/source/libs/index/inc/index_tfile.h index 4928e01a6322ae487c00580d935ba37d9f0c00fd..4618a39197d1d87b1e9d50f895578e9d2dd8fb85 100644 --- a/source/libs/index/inc/index_tfile.h +++ b/source/libs/index/inc/index_tfile.h @@ -49,13 +49,6 @@ typedef struct TFileValue { int32_t offset; } TFileValue; -typedef struct TFileCacheKey { - uint64_t suid; - uint8_t colType; - char* colName; - int32_t nColName; -} TFileCacheKey; - // table cache // refactor to LRU cache later typedef struct TFileCache { @@ -103,10 +96,10 @@ typedef struct TFileReaderOpt { // tfile cache, manage tindex reader TFileCache* tfileCacheCreate(const char* path); void tfileCacheDestroy(TFileCache* tcache); -TFileReader* tfileCacheGet(TFileCache* tcache, TFileCacheKey* key); -void tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader* reader); +TFileReader* tfileCacheGet(TFileCache* tcache, ICacheKey* key); +void tfileCachePut(TFileCache* tcache, ICacheKey* key, TFileReader* reader); -TFileReader* tfileGetReaderByCol(IndexTFile* tf, char* colName); +TFileReader* tfileGetReaderByCol(IndexTFile* tf, uint64_t suid, char* colName); TFileReader* tfileReaderOpen(char* path, uint64_t suid, int32_t version, const char* colName); TFileReader* tfileReaderCreate(WriterCtx* ctx); @@ -124,6 +117,7 @@ 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, SArray* result); diff --git a/source/libs/index/inc/index_util.h b/source/libs/index/inc/index_util.h index 21c5ca155b3bb252429be2dd6c30a344694b168a..adeb52bb8c989332ce8284d24d2369685c147323 100644 --- a/source/libs/index/inc/index_util.h +++ b/source/libs/index/inc/index_util.h @@ -34,7 +34,7 @@ extern "C" { #define SERIALIZE_VAR_TO_BUF(buf, var, type) \ do { \ type c = var; \ - assert(sizeof(var) == sizeof(type)); \ + assert(sizeof(type) == sizeof(c)); \ memcpy((void*)buf, (void*)&c, sizeof(c)); \ buf += sizeof(c); \ } while (0) diff --git a/source/libs/index/src/index.c b/source/libs/index/src/index.c index 5167196031eb822b5540fa3c7b224e64ac859b0a..9c7320b301058a0c91e59d28dc676f9da29f385b 100644 --- a/source/libs/index/src/index.c +++ b/source/libs/index/src/index.c @@ -17,6 +17,7 @@ #include "indexInt.h" #include "index_cache.h" #include "index_tfile.h" +#include "index_util.h" #include "tdef.h" #include "tsched.h" @@ -102,6 +103,7 @@ void indexClose(SIndex* sIdx) { } taosHashCleanup(sIdx->colObj); pthread_mutex_destroy(&sIdx->mtx); + indexTFileDestroy(sIdx->tindex); #endif free(sIdx->path); free(sIdx); @@ -130,18 +132,28 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) { // TODO(yihao): reduce the lock range pthread_mutex_lock(&index->mtx); for (int i = 0; i < taosArrayGetSize(fVals); i++) { - SIndexTerm* p = taosArrayGetP(fVals, i); - IndexCache** cache = taosHashGet(index->colObj, p->colName, p->nColName); + SIndexTerm* p = taosArrayGetP(fVals, i); + + char buf[128] = {0}; + ICacheKey key = {.suid = p->suid, .colName = p->colName}; + int32_t sz = indexSerialCacheKey(&key, buf); + + IndexCache** cache = taosHashGet(index->colObj, buf, sz); if (cache == NULL) { - IndexCache* pCache = indexCacheCreate(index, p->colName, p->colType); - taosHashPut(index->colObj, p->colName, p->nColName, &pCache, sizeof(void*)); + IndexCache* pCache = indexCacheCreate(index, p->suid, p->colName, p->colType); + taosHashPut(index->colObj, buf, sz, &pCache, sizeof(void*)); } } pthread_mutex_unlock(&index->mtx); for (int i = 0; i < taosArrayGetSize(fVals); i++) { - SIndexTerm* p = taosArrayGetP(fVals, i); - IndexCache** cache = taosHashGet(index->colObj, p->colName, p->nColName); + SIndexTerm* p = taosArrayGetP(fVals, i); + + char buf[128] = {0}; + ICacheKey key = {.suid = p->suid, .colName = p->colName}; + int32_t sz = indexSerialCacheKey(&key, buf); + + IndexCache** cache = taosHashGet(index->colObj, buf, sz); assert(*cache != NULL); int ret = indexCachePut(*cache, p, uid); if (ret != 0) { return ret; } @@ -296,7 +308,12 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result // Get col info IndexCache* cache = NULL; pthread_mutex_lock(&sIdx->mtx); - IndexCache** pCache = taosHashGet(sIdx->colObj, colName, nColName); + + char buf[128] = {0}; + ICacheKey key = {.suid = term->suid, .colName = term->colName}; + int32_t sz = indexSerialCacheKey(&key, buf); + + IndexCache** pCache = taosHashGet(sIdx->colObj, buf, sz); if (pCache == NULL) { pthread_mutex_unlock(&sIdx->mtx); return -1; @@ -360,6 +377,7 @@ static void indexMergeSameKey(SArray* result, TFileValue* tv) { if (sz > 0) { // TODO(yihao): remove duplicate tableid TFileValue* lv = taosArrayGetP(result, sz - 1); + // indexError("merge colVal: %s", lv->colVal); if (strcmp(lv->colVal, tv->colVal) == 0) { taosArrayAddAll(lv->tableId, tv->tableId); tfileValueDestroy(tv); @@ -368,6 +386,7 @@ static void indexMergeSameKey(SArray* result, TFileValue* tv) { } } else { taosArrayPush(result, &tv); + // indexError("merge colVal: %s", tv->colVal); } } static void indexDestroyTempResult(SArray* result) { @@ -383,10 +402,12 @@ int indexFlushCacheTFile(SIndex* sIdx, void* cache) { indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid); IndexCache* pCache = (IndexCache*)cache; - TFileReader* pReader = tfileGetReaderByCol(sIdx->tindex, pCache->colName); + TFileReader* pReader = tfileGetReaderByCol(sIdx->tindex, pCache->suid, pCache->colName); + if (pReader == NULL) { indexWarn("empty pReader found"); } // handle flush Iterate* cacheIter = indexCacheIteratorCreate(pCache); Iterate* tfileIter = tfileIteratorCreate(pReader); + if (tfileIter == NULL) { indexWarn("empty tfile reader iterator"); } SArray* result = taosArrayInit(1024, sizeof(void*)); @@ -459,14 +480,14 @@ void iterateValueDestroy(IterateValue* value, bool destroy) { } else { if (value->val != NULL) { taosArrayClear(value->val); } } - // free(value->colVal); + free(value->colVal); value->colVal = NULL; } static int indexGenTFile(SIndex* sIdx, IndexCache* cache, SArray* batch) { int32_t version = CACHE_VERSION(cache); uint8_t colType = cache->type; - TFileWriter* tw = tfileWriterOpen(sIdx->path, sIdx->suid, version, cache->colName, colType); + TFileWriter* tw = tfileWriterOpen(sIdx->path, cache->suid, version, cache->colName, colType); if (tw == NULL) { indexError("failed to open file to write"); return -1; @@ -479,14 +500,13 @@ static int indexGenTFile(SIndex* sIdx, IndexCache* cache, SArray* batch) { } tfileWriterClose(tw); - TFileReader* reader = tfileReaderOpen(sIdx->path, sIdx->suid, version, cache->colName); + TFileReader* reader = tfileReaderOpen(sIdx->path, cache->suid, version, cache->colName); + + char buf[128] = {0}; + TFileHeader* header = &reader->header; + ICacheKey key = { + .suid = cache->suid, .colName = header->colName, .nColName = strlen(header->colName), .colType = header->colType}; - char buf[128] = {0}; - TFileHeader* header = &reader->header; - TFileCacheKey key = {.suid = header->suid, - .colName = header->colName, - .nColName = strlen(header->colName), - .colType = header->colType}; pthread_mutex_lock(&sIdx->mtx); IndexTFile* ifile = (IndexTFile*)sIdx->tindex; @@ -497,3 +517,13 @@ static int indexGenTFile(SIndex* sIdx, IndexCache* cache, SArray* batch) { END: tfileWriterClose(tw); } + +int32_t indexSerialCacheKey(ICacheKey* key, char* buf) { + char* p = buf; + SERIALIZE_MEM_TO_BUF(buf, key, suid); + SERIALIZE_VAR_TO_BUF(buf, '_', char); + // SERIALIZE_MEM_TO_BUF(buf, key, colType); + // SERIALIZE_VAR_TO_BUF(buf, '_', char); + SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName); + return buf - p; +} diff --git a/source/libs/index/src/index_cache.c b/source/libs/index/src/index_cache.c index 0e46445a00db0a2bcc4c40b03b4b6ea95fa64e52..b4c533e998647c6540da8eef35604a4f05922377 100644 --- a/source/libs/index/src/index_cache.c +++ b/source/libs/index/src/index_cache.c @@ -20,7 +20,7 @@ #define MAX_INDEX_KEY_LEN 256 // test only, change later -#define MEM_TERM_LIMIT 10000 * 10 +#define MEM_TERM_LIMIT 10 * 10000 // ref index_cache.h:22 //#define CACHE_KEY_LEN(p) \ // (sizeof(int32_t) + sizeof(uint16_t) + sizeof(p->colType) + sizeof(p->nColVal) + p->nColVal + sizeof(uint64_t) + @@ -40,7 +40,7 @@ static bool indexCacheIteratorNext(Iterate* itera); static IterateValue* indexCacheIteratorGetValue(Iterate* iter); -IndexCache* indexCacheCreate(SIndex* idx, const char* colName, int8_t type) { +IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type) { IndexCache* cache = calloc(1, sizeof(IndexCache)); if (cache == NULL) { indexError("failed to create index cache"); @@ -53,7 +53,7 @@ IndexCache* indexCacheCreate(SIndex* idx, const char* colName, int8_t type) { cache->type = type; cache->index = idx; cache->version = 0; - + cache->suid = suid; pthread_mutex_init(&cache->mtx, NULL); indexCacheRef(cache); return cache; @@ -150,6 +150,7 @@ Iterate* indexCacheIteratorCreate(IndexCache* cache) { MemTable* tbl = cache->imm; iiter->val.val = taosArrayInit(1, sizeof(uint64_t)); + iiter->val.colVal = NULL; iiter->iter = tbl != NULL ? tSkipListCreateIter(tbl->mem) : NULL; iiter->next = indexCacheIteratorNext; iiter->getValue = indexCacheIteratorGetValue; @@ -353,6 +354,9 @@ static bool indexCacheIteratorNext(Iterate* itera) { SSkipListIterator* iter = itera->iter; if (iter == NULL) { return false; } IterateValue* iv = &itera->val; + if (iv->colVal != NULL && iv->val != NULL) { + // indexError("value in cache: colVal: %s, size: %d", iv->colVal, (int)taosArrayGetSize(iv->val)); + } iterateValueDestroy(iv, false); bool next = tSkipListIterNext(iter); diff --git a/source/libs/index/src/index_fst.c b/source/libs/index/src/index_fst.c index 04a08dafd2641f7fd1f91d660d8c4ff4013abeb2..bfaeeaaa33425c8c51f832e8c2512f4906cedffa 100644 --- a/source/libs/index/src/index_fst.c +++ b/source/libs/index/src/index_fst.c @@ -319,7 +319,7 @@ void fstStateSetCommInput(FstState* s, uint8_t inp) { assert(s->state == OneTransNext || s->state == OneTrans); uint8_t val; - COMMON_INDEX(inp, 0x111111, val); + COMMON_INDEX(inp, 0b111111, val); s->val = (s->val & fstStateDict[s->state].val) | val; } @@ -369,7 +369,7 @@ uint8_t fstStateInput(FstState* s, FstNode* node) { bool null = false; uint8_t inp = fstStateCommInput(s, &null); uint8_t* data = fstSliceData(slice, NULL); - return null == false ? inp : data[-1]; + return null == false ? inp : data[node->start - 1]; } uint8_t fstStateInputForAnyTrans(FstState* s, FstNode* node, uint64_t i) { assert(s->state == AnyTrans); @@ -1062,6 +1062,7 @@ Output fstEmptyFinalOutput(Fst* fst, bool* null) { } else { *null = true; } + fstNodeDestroy(node); return res; } @@ -1286,6 +1287,7 @@ StreamWithStateResult* streamWithStateNextWith(StreamWithState* sws, StreamCallb StreamWithStateResult* result = swsResultCreate(&slice, fOutput, tState); free(buf); fstSliceDestroy(&slice); + taosArrayDestroy(nodes); return result; } free(buf); diff --git a/source/libs/index/src/index_tfile.c b/source/libs/index/src/index_tfile.c index fc4f8593a105f5898fb977a8de0110bf83e5d6e5..95c713fb0a5a328c50761776db76725da2fa0229 100644 --- a/source/libs/index/src/index_tfile.c +++ b/source/libs/index/src/index_tfile.c @@ -51,7 +51,6 @@ static void tfileDestroyFileName(void* elem); static int tfileCompare(const void* a, const void* b); static int tfileParseFileName(const char* filename, uint64_t* suid, int* colId, int* version); static void tfileGenFileName(char* filename, uint64_t suid, int colId, int version); -static void tfileSerialCacheKey(TFileCacheKey* key, char* buf); TFileCache* tfileCacheCreate(const char* path) { TFileCache* tcache = calloc(1, sizeof(TFileCache)); @@ -80,18 +79,18 @@ TFileCache* tfileCacheCreate(const char* path) { goto End; } - char buf[128] = {0}; - TFileReader* reader = tfileReaderCreate(wc); - TFileHeader* header = &reader->header; - TFileCacheKey key = {.suid = header->suid, - .colName = header->colName, - .nColName = strlen(header->colName), - .colType = header->colType}; - tfileSerialCacheKey(&key, buf); - + char buf[128] = {0}; + TFileReader* reader = tfileReaderCreate(wc); + TFileHeader* header = &reader->header; + ICacheKey key = {.suid = header->suid, + .colName = header->colName, + .nColName = strlen(header->colName), + .colType = header->colType}; + + int32_t sz = indexSerialCacheKey(&key, buf); + assert(sz < sizeof(buf)); + taosHashPut(tcache->tableCache, buf, sz, &reader, sizeof(void*)); tfileReaderRef(reader); - // indexTable - taosHashPut(tcache->tableCache, buf, strlen(buf), &reader, sizeof(void*)); } taosArrayDestroyEx(files, tfileDestroyFileName); return tcache; @@ -117,30 +116,30 @@ void tfileCacheDestroy(TFileCache* tcache) { free(tcache); } -TFileReader* tfileCacheGet(TFileCache* tcache, TFileCacheKey* key) { - char buf[128] = {0}; - tfileSerialCacheKey(key, buf); - - TFileReader** reader = taosHashGet(tcache->tableCache, buf, strlen(buf)); +TFileReader* tfileCacheGet(TFileCache* tcache, ICacheKey* key) { + char buf[128] = {0}; + int32_t sz = indexSerialCacheKey(key, buf); + assert(sz < sizeof(buf)); + TFileReader** reader = taosHashGet(tcache->tableCache, buf, sz); if (reader == NULL) { return NULL; } tfileReaderRef(*reader); return *reader; } -void tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader* reader) { - char buf[128] = {0}; - tfileSerialCacheKey(key, buf); +void tfileCachePut(TFileCache* tcache, ICacheKey* key, TFileReader* reader) { + char buf[128] = {0}; + int32_t sz = indexSerialCacheKey(key, buf); // remove last version index reader - TFileReader** p = taosHashGet(tcache->tableCache, buf, strlen(buf)); + TFileReader** p = taosHashGet(tcache->tableCache, buf, sz); if (p != NULL) { TFileReader* oldReader = *p; - taosHashRemove(tcache->tableCache, buf, strlen(buf)); + taosHashRemove(tcache->tableCache, buf, sz); oldReader->remove = true; tfileReaderUnRef(oldReader); } + taosHashPut(tcache->tableCache, buf, sz, &reader, sizeof(void*)); tfileReaderRef(reader); - taosHashPut(tcache->tableCache, buf, strlen(buf), &reader, sizeof(void*)); return; } TFileReader* tfileReaderCreate(WriterCtx* ctx) { @@ -230,8 +229,6 @@ TFileReader* tfileReaderOpen(char* path, uint64_t suid, int32_t version, const c TFileReader* reader = tfileReaderCreate(wc); return reader; - - // tfileSerialCacheKey(&key, buf); } TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header) { // char pathBuf[128] = {0}; @@ -325,15 +322,19 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) { tfileWriterClose(tw); return -1; } - // write fst + + // write data indexError("--------Begin----------------"); for (size_t i = 0; i < sz; i++) { // TODO, fst batch write later TFileValue* v = taosArrayGetP((SArray*)data, i); - if (tfileWriteData(tw, v) == 0) { - // + if (tfileWriteData(tw, v) != 0) { + indexError("failed to write data: %s, offset: %d len: %d", v->colVal, v->offset, + (int)taosArrayGetSize(v->tableId)); + } else { + indexInfo("success to write data: %s, offset: %d len: %d", v->colVal, v->offset, + (int)taosArrayGetSize(v->tableId)); } - indexError("data: %s, offset: %d len: %d", v->colVal, v->offset, (int)taosArrayGetSize(v->tableId)); } indexError("--------End----------------"); fstBuilderFinish(tw->fb); @@ -359,7 +360,7 @@ IndexTFile* indexTFileCreate(const char* path) { tfile->cache = tfileCacheCreate(path); return tfile; } -void IndexTFileDestroy(IndexTFile* tfile) { +void indexTFileDestroy(IndexTFile* tfile) { tfileCacheDestroy(tfile->cache); free(tfile); } @@ -369,9 +370,8 @@ int indexTFileSearch(void* tfile, SIndexTermQuery* query, SArray* result) { if (tfile == NULL) { return ret; } IndexTFile* pTfile = (IndexTFile*)tfile; - SIndexTerm* term = query->term; - TFileCacheKey key = { - .suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName}; + SIndexTerm* term = query->term; + ICacheKey key = {.suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName}; TFileReader* reader = tfileCacheGet(pTfile->cache, &key); if (reader == NULL) { return 0; } @@ -385,8 +385,10 @@ int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid) { } static bool tfileIteratorNext(Iterate* iiter) { IterateValue* iv = &iiter->val; + if (iv->colVal != NULL && iv->val != NULL) { + // indexError("value in fst: colVal: %s, size: %d", iv->colVal, (int)taosArrayGetSize(iv->val)); + } iterateValueDestroy(iv, false); - // SArray* tblIds = iv->val; char* colVal = NULL; uint64_t offset = 0; @@ -406,14 +408,14 @@ static bool tfileIteratorNext(Iterate* iiter) { if (tfileReaderLoadTableIds(tIter->rdr, offset, iv->val) != 0) { return false; } iv->colVal = colVal; - + return true; // std::string key(ch, sz); } static IterateValue* tifileIterateGetValue(Iterate* iter) { return &iter->val; } static TFileFstIter* tfileFstIteratorCreate(TFileReader* reader) { - TFileFstIter* tIter = calloc(1, sizeof(Iterate)); + TFileFstIter* tIter = calloc(1, sizeof(TFileFstIter)); if (tIter == NULL) { return NULL; } tIter->ctx = automCtxCreate(NULL, AUTOMATION_ALWAYS); @@ -435,6 +437,7 @@ Iterate* tfileIteratorCreate(TFileReader* reader) { iter->next = tfileIteratorNext; iter->getValue = tifileIterateGetValue; iter->val.val = taosArrayInit(1, sizeof(uint64_t)); + iter->val.colVal = NULL; return iter; } void tfileIteratorDestroy(Iterate* iter) { @@ -447,13 +450,14 @@ void tfileIteratorDestroy(Iterate* iter) { streamWithStateDestroy(tIter->st); fstStreamBuilderDestroy(tIter->fb); automCtxDestroy(tIter->ctx); + free(tIter); free(iter); } -TFileReader* tfileGetReaderByCol(IndexTFile* tf, char* colName) { +TFileReader* tfileGetReaderByCol(IndexTFile* tf, uint64_t suid, char* colName) { if (tf == NULL) { return NULL; } - TFileCacheKey key = {.suid = 0, .colType = TSDB_DATA_TYPE_BINARY, .colName = colName, .nColName = strlen(colName)}; + ICacheKey key = {.suid = suid, .colType = TSDB_DATA_TYPE_BINARY, .colName = colName, .nColName = strlen(colName)}; return tfileCacheGet(tf->cache, &key); } @@ -480,7 +484,7 @@ static int tfileValueCompare(const void* a, const void* b, const void* param) { TFileValue* tfileValueCreate(char* val) { TFileValue* tf = calloc(1, sizeof(TFileValue)); if (tf == NULL) { return NULL; } - tf->colVal = val; + tf->colVal = tstrdup(val); tf->tableId = taosArrayInit(32, sizeof(uint64_t)); return tf; } @@ -491,6 +495,7 @@ int tfileValuePush(TFileValue* tf, uint64_t val) { } void tfileValueDestroy(TFileValue* tf) { taosArrayDestroy(tf->tableId); + free(tf->colVal); free(tf); } static void tfileSerialTableIdsToBuf(char* buf, SArray* ids) { @@ -648,10 +653,3 @@ static int tfileParseFileName(const char* filename, uint64_t* suid, int* colId, } return -1; } -static void tfileSerialCacheKey(TFileCacheKey* key, char* buf) { - // SERIALIZE_MEM_TO_BUF(buf, key, suid); - // SERIALIZE_VAR_TO_BUF(buf, '_', char); - // SERIALIZE_MEM_TO_BUF(buf, key, colType); - // SERIALIZE_VAR_TO_BUF(buf, '_', char); - SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName); -} diff --git a/source/libs/index/test/fstTest.cc b/source/libs/index/test/fstTest.cc index da974ce6c4be500370eefe4b112f507b5e7912e6..3d978c05a5725e8f6551e1b3e15a0bffc2353663 100644 --- a/source/libs/index/test/fstTest.cc +++ b/source/libs/index/test/fstTest.cc @@ -24,8 +24,13 @@ class FstWriter { _b = fstBuilderCreate(_wc, 0); } bool Put(const std::string& key, uint64_t val) { + // char buf[128] = {0}; + // int len = 0; + // taosMbsToUcs4(key.c_str(), key.size(), buf, 128, &len); + // FstSlice skey = fstSliceCreate((uint8_t*)buf, len); FstSlice skey = fstSliceCreate((uint8_t*)key.c_str(), key.size()); bool ok = fstBuilderInsert(_b, skey, val); + fstSliceDestroy(&skey); return ok; } @@ -61,6 +66,11 @@ class FstReadMemory { return _fst != NULL; } bool Get(const std::string& key, uint64_t* val) { + // char buf[128] = {0}; + // int len = 0; + // taosMbsToUcs4(key.c_str(), key.size(), buf, 128, &len); + // FstSlice skey = fstSliceCreate((uint8_t*)buf, len); + FstSlice skey = fstSliceCreate((uint8_t*)key.c_str(), key.size()); bool ok = fstGet(_fst, &skey, val); fstSliceDestroy(&skey); @@ -135,15 +145,109 @@ int Performance_fstWriteRecords(FstWriter* b) { } return L * M * N; } +void Performance_fstReadRecords(FstReadMemory* m) { + std::string str("aa"); + for (int i = 0; i < M; i++) { + str[0] = 'a' + i; + str.resize(2); + for (int j = 0; j < N; j++) { + str[1] = 'a' + j; + str.resize(2); + for (int k = 0; k < L; k++) { + str.push_back('a'); + uint64_t val, cost; + if (m->GetWithTimeCostUs(str, &val, &cost)) { + printf("succes to get kv(%s, %" PRId64 "), cost: %" PRId64 "\n", str.c_str(), val, cost); + } else { + printf("failed to get key: %s\n", str.c_str()); + } + } + } + } +} + +void checkMillonWriteAndReadOfFst() { + tfInit(); + FstWriter* fw = new FstWriter; + Performance_fstWriteRecords(fw); + delete fw; + FstReadMemory* fr = new FstReadMemory(1024 * 64 * 1024); + + if (fr->init()) { printf("success to init fst read"); } + + Performance_fstReadRecords(fr); + tfCleanup(); + delete fr; +} +void checkFstLongTerm() { + tfInit(); + FstWriter* fw = new FstWriter; + // Performance_fstWriteRecords(fw); + + fw->Put("A B", 1); + fw->Put("C", 2); + fw->Put("a", 3); + delete fw; + + FstReadMemory* m = new FstReadMemory(1024 * 64); + if (m->init() == false) { + std::cout << "init readMemory failed" << std::endl; + delete m; + return; + } + { + uint64_t val = 0; + if (m->Get("A B", &val)) { + std::cout << "success to Get: " << val << std::endl; + } else { + std::cout << "failed to Get:" << val << std::endl; + } + } + { + uint64_t val = 0; + if (m->Get("C", &val)) { + std::cout << "success to Get: " << val << std::endl; + } else { + std::cout << "failed to Get:" << val << std::endl; + } + } + { + uint64_t val = 0; + if (m->Get("a", &val)) { + std::cout << "success to Get: " << val << std::endl; + } else { + std::cout << "failed to Get:" << val << std::endl; + } + } + + // prefix search + // std::vector result; + + // AutomationCtx* ctx = automCtxCreate((void*)"ab", AUTOMATION_ALWAYS); + // m->Search(ctx, result); + // std::cout << "size: " << result.size() << std::endl; + // assert(result.size() == count); + // for (int i = 0; i < result.size(); i++) { + // assert(result[i] == i); // check result + //} + tfCleanup(); + // free(ctx); + // delete m; +} void checkFstCheckIterator() { tfInit(); FstWriter* fw = new FstWriter; int64_t s = taosGetTimestampUs(); int count = 2; - Performance_fstWriteRecords(fw); + // Performance_fstWriteRecords(fw); int64_t e = taosGetTimestampUs(); std::cout << "insert data count : " << count << "elapas time: " << e - s << std::endl; + + fw->Put("Hello world", 1); + fw->Put("hello world", 2); + fw->Put("hello worle", 3); + fw->Put("hello worlf", 4); delete fw; FstReadMemory* m = new FstReadMemory(1024 * 64); @@ -171,7 +275,7 @@ void checkFstCheckIterator() { void fst_get(Fst* fst) { for (int i = 0; i < 10000; i++) { - std::string term = "Hello"; + std::string term = "Hello World"; FstSlice key = fstSliceCreate((uint8_t*)term.c_str(), term.size()); uint64_t offset = 0; bool ret = fstGet(fst, &key, &offset); @@ -189,7 +293,7 @@ void validateTFile(char* arg) { std::thread threads[NUM_OF_THREAD]; // std::vector threads; - TFileReader* reader = tfileReaderOpen(arg, 0, 295868, "tag1"); + TFileReader* reader = tfileReaderOpen(arg, 0, 999992, "tag1"); for (int i = 0; i < NUM_OF_THREAD; i++) { threads[i] = std::thread(fst_get, reader->fst); @@ -203,9 +307,12 @@ void validateTFile(char* arg) { tfCleanup(); } int main(int argc, char* argv[]) { - if (argc > 1) { validateTFile(argv[1]); } + // tool to check all kind of fst test + // if (argc > 1) { validateTFile(argv[1]); } // checkFstCheckIterator(); + // checkFstLongTerm(); // checkFstPrefixSearch(); + checkMillonWriteAndReadOfFst(); return 1; } diff --git a/source/libs/index/test/indexTests.cc b/source/libs/index/test/indexTests.cc index 080becccf183cd374ed833895c02b1f5a949803f..bdfb86ce1714ab3b309091718c1b72676b8120fd 100644 --- a/source/libs/index/test/indexTests.cc +++ b/source/libs/index/test/indexTests.cc @@ -457,7 +457,10 @@ TEST_F(IndexTFileEnv, test_tfile_write) { // taosArrayPush(data, &v4); fObj->Put(data); - for (size_t i = 0; i < taosArrayGetSize(data); i++) { destroyTFileValue(taosArrayGetP(data, i)); } + for (size_t i = 0; i < taosArrayGetSize(data); i++) { + // data + destroyTFileValue(taosArrayGetP(data, i)); + } taosArrayDestroy(data); std::string colName("voltage"); @@ -470,6 +473,7 @@ TEST_F(IndexTFileEnv, test_tfile_write) { fObj->Get(&query, result); assert(taosArrayGetSize(result) == 200); indexTermDestroy(term); + taosArrayDestroy(result); // tfileWriterDestroy(twrite); } @@ -477,7 +481,7 @@ class CacheObj { public: CacheObj() { // TODO - cache = indexCacheCreate(NULL, "voltage", TSDB_DATA_TYPE_BINARY); + cache = indexCacheCreate(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); @@ -534,6 +538,7 @@ TEST_F(IndexCacheEnv, cache_test) { SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); + indexTermDestroy(term); // indexTermDestry(term); } { @@ -541,24 +546,28 @@ TEST_F(IndexCacheEnv, cache_test) { SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); + indexTermDestroy(term); } { std::string colVal("v2"); SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); + indexTermDestroy(term); } { std::string colVal("v3"); SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); + indexTermDestroy(term); } { std::string colVal("v3"); SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); + indexTermDestroy(term); } coj->Debug(); std::cout << "--------first----------" << std::endl; @@ -567,12 +576,14 @@ TEST_F(IndexCacheEnv, cache_test) { SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, othColId, version++, suid++); + indexTermDestroy(term); } { std::string colVal("v4"); SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, othColId, version++, suid++); + indexTermDestroy(term); } coj->Debug(); std::cout << "--------second----------" << std::endl; @@ -583,6 +594,7 @@ TEST_F(IndexCacheEnv, cache_test) { SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); + indexTermDestroy(term); } } coj->Debug(); @@ -598,6 +610,9 @@ TEST_F(IndexCacheEnv, cache_test) { coj->Get(&query, colId, 10000, ret, &valType); std::cout << "size : " << taosArrayGetSize(ret) << std::endl; assert(taosArrayGetSize(ret) == 4); + taosArrayDestroy(ret); + + indexTermDestroy(term); } { std::string colVal("v2"); @@ -609,6 +624,9 @@ TEST_F(IndexCacheEnv, cache_test) { coj->Get(&query, colId, 10000, ret, &valType); assert(taosArrayGetSize(ret) == 1); + taosArrayDestroy(ret); + + indexTermDestroy(term); } } class IndexObj { @@ -678,13 +696,16 @@ class IndexObj { SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t)); if (Search(mq, result) == 0) { std::cout << "search one successfully" << std::endl; } - return taosArrayGetSize(result); + int sz = taosArrayGetSize(result); + indexMultiTermQueryDestroy(mq); + taosArrayDestroy(result); + return sz; // assert(taosArrayGetSize(result) == targetSize); } void PutOne(const std::string& colName, const std::string& colVal) { + SIndexMultiTerm* terms = indexMultiTermCreate(); SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); - SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); Put(terms, 10); indexMultiTermDestroy(terms); @@ -783,18 +804,21 @@ TEST_F(IndexEnv2, testIndexOpen) { index->Search(mq, result); std::cout << "target size: " << taosArrayGetSize(result) << std::endl; assert(taosArrayGetSize(result) == 400); + taosArrayDestroy(result); + indexMultiTermQueryDestroy(mq); } } TEST_F(IndexEnv2, testIndex_TrigeFlush) { - std::string path = "/tmp/test"; + std::string path = "/tmp/test1"; if (index->Init(path) != 0) { // r std::cout << "failed to init" << std::endl; } int numOfTable = 100 * 10000; - index->WriteMillonData("tag1", "Hello", numOfTable); - int target = index->SearchOne("tag1", "Hello"); + index->WriteMillonData("tag1", "Hello Wolrd", numOfTable); + int target = index->SearchOne("tag1", "Hello Wolrd"); + std::cout << "Get Index: " << target << std::endl; assert(numOfTable == target); } @@ -821,14 +845,6 @@ TEST_F(IndexEnv2, testIndex_serarch_cache_and_tfile) { threads[i].join(); } } -TEST_F(IndexEnv2, testIndex_multi_thread_write) { - std::string path = "/tmp"; - if (index->Init(path) != 0) {} -} -TEST_F(IndexEnv2, testIndex_multi_thread_read) { - std::string path = "/tmp"; - if (index->Init(path) != 0) {} -} TEST_F(IndexEnv2, testIndex_restart) { std::string path = "/tmp";