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

feat: add lru to index

This commit is contained in:
yihaoDeng 2022-07-14 14:51:39 +08:00
parent 88ed8d4cf5
commit cace9b6be9
13 changed files with 220 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/index/index.h
View File

@ -173,7 +173,7 @@ void indexMultiTermDestroy(SIndexMultiTerm* terms);
* @param:
* @param:
*/
SIndexOpts* indexOptsCreate();
SIndexOpts* indexOptsCreate(int32_t cacheSize);
void indexOptsDestroy(SIndexOpts* opts);
/*

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

@ -99,12 +99,12 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
goto _err;
}
// open pTagIdx
// TODO(yihaoDeng), refactor later
char indexFullPath[128] = {0};
sprintf(indexFullPath, "%s/%s", pMeta->path, "invert");
taosMkDir(indexFullPath);
ret = indexOpen(indexOptsCreate(), indexFullPath, (SIndex **)&pMeta->pTagIvtIdx);
SIndexOpts *opts = indexOptsCreate(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;

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

@ -27,7 +27,7 @@ extern "C" {
#define DefaultMem 1024 * 1024
static char tmpFile[] = "./index";
typedef enum WriterType { TMemory, TFile } WriterType;
typedef enum WriterType { TMEMORY, TFILE } WriterType;
typedef struct IFileCtx {
int (*write)(struct IFileCtx* ctx, uint8_t* buf, int len);
@ -35,6 +35,8 @@ typedef struct IFileCtx {
int (*flush)(struct IFileCtx* ctx);
int (*readFrom)(struct IFileCtx* ctx, uint8_t* buf, int len, int32_t offset);
int (*size)(struct IFileCtx* ctx);
SLRUCache* lru;
WriterType type;
union {
struct {

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

@ -24,12 +24,9 @@
#include "tchecksum.h"
#include "thash.h"
#include "tlog.h"
#include "tlrucache.h"
#include "tutil.h"
#ifdef USE_LUCENE
#include <lucene++/Lucene_c.h>
#endif
#ifdef __cplusplus
extern "C" {
#endif
@ -61,28 +58,22 @@ struct SIndex {
void* tindex;
SHashObj* colObj; // < field name, field id>
int64_t suid; // current super table id, -1 is normal table
int32_t cVersion; // current version allocated to cache
char* path;
int64_t suid; // current super table id, -1 is normal table
int32_t cVersion; // current version allocated to cache
SLRUCache* lru;
char* path;
int8_t status;
SIndexStat stat;
TdThreadMutex mtx;
tsem_t sem;
bool quit;
void* opts;
};
struct SIndexOpts {
#ifdef USE_LUCENE
void* opts;
#endif
#ifdef USE_INVERTED_INDEX
int32_t cacheSize; // MB
// add cache module later
#endif
int32_t cacheOpt; // MB
int32_t cacheOpt; // MB
};
struct SIndexMultiTermQuery {

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

@ -71,6 +71,7 @@ typedef struct TFileReader {
IFileCtx* ctx;
TFileHeader header;
bool remove;
void* lru;
} TFileReader;
typedef struct IndexTFile {
@ -95,14 +96,14 @@ typedef struct TFileReaderOpt {
} TFileReaderOpt;
// tfile cache, manage tindex reader
TFileCache* tfileCacheCreate(const char* path);
TFileCache* tfileCacheCreate(SIndex* idx, const char* path);
void tfileCacheDestroy(TFileCache* tcache);
TFileReader* tfileCacheGet(TFileCache* tcache, ICacheKey* key);
void tfileCachePut(TFileCache* tcache, ICacheKey* key, TFileReader* reader);
TFileReader* tfileGetReaderByCol(IndexTFile* tf, uint64_t suid, char* colName);
TFileReader* tfileReaderOpen(char* path, uint64_t suid, int64_t version, const char* colName);
TFileReader* tfileReaderOpen(SIndex* idx, uint64_t suid, int64_t version, const char* colName);
TFileReader* tfileReaderCreate(IFileCtx* ctx);
void tfileReaderDestroy(TFileReader* reader);
int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTRslt* tr);
@ -117,7 +118,7 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order);
int tfileWriterFinish(TFileWriter* tw);
//
IndexTFile* idxTFileCreate(const char* path);
IndexTFile* idxTFileCreate(SIndex* idx, const char* path);
void idxTFileDestroy(IndexTFile* tfile);
int idxTFilePut(void* tfile, SIndexTerm* term, uint64_t uid);
int idxTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* tr);

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

@ -103,44 +103,61 @@ static void indexWait(void* idx) {
int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
int ret = TSDB_CODE_SUCCESS;
taosThreadOnce(&isInit, indexInit);
SIndex* sIdx = taosMemoryCalloc(1, sizeof(SIndex));
if (sIdx == NULL) {
SIndex* idx = taosMemoryCalloc(1, sizeof(SIndex));
if (idx == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
sIdx->tindex = idxTFileCreate(path);
if (sIdx->tindex == NULL) {
idx->lru = taosLRUCacheInit(opts->cacheSize, -1, .5);
if (idx->lru == NULL) {
ret = TSDB_CODE_OUT_OF_MEMORY;
goto END;
}
taosLRUCacheSetStrictCapacity(idx->lru, true);
idx->tindex = idxTFileCreate(idx, path);
if (idx->tindex == NULL) {
ret = TSDB_CODE_OUT_OF_MEMORY;
goto END;
}
sIdx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
sIdx->cVersion = 1;
sIdx->path = tstrdup(path);
taosThreadMutexInit(&sIdx->mtx, NULL);
tsem_init(&sIdx->sem, 0, 0);
idx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
idx->cVersion = 1;
idx->path = tstrdup(path);
taosThreadMutexInit(&idx->mtx, NULL);
tsem_init(&idx->sem, 0, 0);
sIdx->refId = idxAddRef(sIdx);
idxAcquireRef(sIdx->refId);
idx->refId = idxAddRef(idx);
idx->opts = opts;
idxAcquireRef(idx->refId);
*index = sIdx;
*index = idx;
return ret;
END:
if (sIdx != NULL) {
indexClose(sIdx);
if (idx != NULL) {
indexClose(idx);
}
*index = NULL;
return ret;
}
void indexDestroy(void* handle) {
SIndex* sIdx = handle;
taosThreadMutexDestroy(&sIdx->mtx);
tsem_destroy(&sIdx->sem);
idxTFileDestroy(sIdx->tindex);
taosMemoryFree(sIdx->path);
taosMemoryFree(sIdx);
SIndex* idx = handle;
taosThreadMutexDestroy(&idx->mtx);
tsem_destroy(&idx->sem);
idxTFileDestroy(idx->tindex);
taosMemoryFree(idx->path);
SLRUCache* lru = idx->lru;
if (lru != NULL) {
taosLRUCacheEraseUnrefEntries(lru);
taosLRUCacheCleanup(lru);
}
idx->lru = NULL;
indexOptsDestroy(idx->opts);
taosMemoryFree(idx);
return;
}
void indexClose(SIndex* sIdx) {
@ -159,6 +176,7 @@ void indexClose(SIndex* sIdx) {
taosHashCleanup(sIdx->colObj);
sIdx->colObj = NULL;
}
idxReleaseRef(sIdx->refId);
idxRemoveRef(sIdx->refId);
}
@ -234,8 +252,12 @@ int indexSearch(SIndex* index, SIndexMultiTermQuery* multiQuerys, SArray* result
int indexDelete(SIndex* index, SIndexMultiTermQuery* query) { return 1; }
// int indexRebuild(SIndex* index, SIndexOpts* opts) { return 0; }
SIndexOpts* indexOptsCreate() { return NULL; }
void indexOptsDestroy(SIndexOpts* opts) { return; }
SIndexOpts* indexOptsCreate(int32_t cacheSize) {
SIndexOpts* opts = taosMemoryCalloc(1, sizeof(SIndexOpts));
opts->cacheSize = cacheSize;
return opts;
}
void indexOptsDestroy(SIndexOpts* opts) { return taosMemoryFree(opts); }
/*
* @param: oper
*
@ -641,7 +663,7 @@ static int idxGenTFile(SIndex* sIdx, IndexCache* cache, SArray* batch) {
}
tfileWriterClose(tw);
TFileReader* reader = tfileReaderOpen(sIdx->path, cache->suid, version, cache->colName);
TFileReader* reader = tfileReaderOpen(sIdx, cache->suid, version, cache->colName);
if (reader == NULL) {
return -1;
}

5
source/libs/index/src/indexFst.c
View File

@ -772,6 +772,7 @@ void fstBuilderDestroy(FstBuilder* b) {
if (b == NULL) {
return;
}
fstBuilderFinish(b);
idxFileDestroy(b->wrt);
fstUnFinishedNodesDestroy(b->unfinished);
@ -1074,8 +1075,8 @@ FStmStBuilder* fstSearchWithState(Fst* fst, FAutoCtx* ctx) {
}
FstNode* fstGetRoot(Fst* fst) {
CompiledAddr rAddr = fstGetRootAddr(fst);
return fstGetNode(fst, rAddr);
CompiledAddr addr = fstGetRootAddr(fst);
return fstGetNode(fst, addr);
}
FstNode* fstGetNode(Fst* fst, CompiledAddr addr) {

100
source/libs/index/src/indexFstFile.c
View File

@ -4,8 +4,7 @@
* 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
* * 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.
*
@ -14,13 +13,32 @@
*/
#include "indexFstFile.h"
#include "indexComm.h"
#include "indexFstUtil.h"
#include "indexInt.h"
#include "indexUtil.h"
#include "os.h"
#include "tutil.h"
static int32_t kBlockSize = 4096;
typedef struct {
int32_t blockId;
int32_t nread;
char buf[0];
} SDataBlock;
static void deleteDataBlockFromLRU(const void* key, size_t keyLen, void* value) { taosMemoryFree(value); }
static void idxGenLRUKey(char* buf, const char* path, int32_t blockId) {
char* p = buf;
SERIALIZE_STR_VAR_TO_BUF(p, path, strlen(path));
SERIALIZE_VAR_TO_BUF(p, '_', char);
idxInt2str(blockId, p, 0);
return;
}
static int idxFileCtxDoWrite(IFileCtx* ctx, uint8_t* buf, int len) {
if (ctx->type == TFile) {
if (ctx->type == TFILE) {
assert(len == taosWriteFile(ctx->file.pFile, buf, len));
} else {
memcpy(ctx->mem.buf + ctx->offset, buf, len);
@ -30,7 +48,7 @@ static int idxFileCtxDoWrite(IFileCtx* ctx, uint8_t* buf, int len) {
}
static int idxFileCtxDoRead(IFileCtx* ctx, uint8_t* buf, int len) {
int nRead = 0;
if (ctx->type == TFile) {
if (ctx->type == TFILE) {
#ifdef USE_MMAP
nRead = len < ctx->file.size ? len : ctx->file.size;
memcpy(buf, ctx->file.ptr, nRead);
@ -45,24 +63,59 @@ static int idxFileCtxDoRead(IFileCtx* ctx, uint8_t* buf, int len) {
return nRead;
}
static int idxFileCtxDoReadFrom(IFileCtx* ctx, uint8_t* buf, int len, int32_t offset) {
int nRead = 0;
if (ctx->type == TFile) {
// tfLseek(ctx->file.pFile, offset, 0);
int32_t total = 0, nread = 0;
int32_t blkId = offset / kBlockSize;
int32_t blkOffset = offset % kBlockSize;
int32_t blkLeft = kBlockSize - blkOffset;
do {
char key[128] = {0};
idxGenLRUKey(key, ctx->file.buf, blkId);
LRUHandle* h = taosLRUCacheLookup(ctx->lru, key, strlen(key));
if (h) {
SDataBlock* blk = taosLRUCacheValue(ctx->lru, h);
nread = MIN(blkLeft, len);
memcpy(buf + total, blk->buf + blkOffset, nread);
taosLRUCacheRelease(ctx->lru, h, false);
} else {
int32_t cacheMemSize = sizeof(SDataBlock) + kBlockSize;
SDataBlock* blk = taosMemoryCalloc(1, cacheMemSize);
blk->blockId = blkId;
blk->nread = taosPReadFile(ctx->file.pFile, blk->buf, kBlockSize, blkId * kBlockSize);
assert(blk->nread <= kBlockSize);
nread = MIN(blkLeft, len);
if (blk->nread < kBlockSize && blk->nread < len) {
break;
}
memcpy(buf + total, blk->buf + blkOffset, nread);
LRUStatus s = taosLRUCacheInsert(ctx->lru, key, strlen(key), blk, cacheMemSize, deleteDataBlockFromLRU, NULL,
TAOS_LRU_PRIORITY_LOW);
if (s != TAOS_LRU_STATUS_OK) {
return -1;
}
}
total += nread;
len -= nread;
offset += nread;
blkId = offset / kBlockSize;
blkOffset = offset % kBlockSize;
blkLeft = kBlockSize - blkOffset;
} while (len > 0);
#ifdef USE_MMAP
int32_t last = ctx->file.size - offset;
nRead = last >= len ? len : last;
memcpy(buf, ctx->file.ptr + offset, nRead);
#else
nRead = taosPReadFile(ctx->file.pFile, buf, len, offset);
int32_t last = ctx->file.size - offset;
nRead = last >= len ? len : last;
memcpy(buf, ctx->file.ptr + offset, nRead);
#endif
} else {
// refactor later
assert(0);
}
return nRead;
return total;
}
static int idxFileCtxGetSize(IFileCtx* ctx) {
if (ctx->type == TFile) {
if (ctx->type == TFILE) {
int64_t file_size = 0;
taosStatFile(ctx->file.buf, &file_size, NULL);
return (int)file_size;
@ -70,7 +123,7 @@ static int idxFileCtxGetSize(IFileCtx* ctx) {
return 0;
}
static int idxFileCtxDoFlush(IFileCtx* ctx) {
if (ctx->type == TFile) {
if (ctx->type == TFILE) {
taosFsyncFile(ctx->file.pFile);
} else {
// do nothing
@ -85,7 +138,7 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int
}
ctx->type = type;
if (ctx->type == TFile) {
if (ctx->type == TFILE) {
// ugly code, refactor later
ctx->file.readOnly = readOnly;
memcpy(ctx->file.buf, path, strlen(path));
@ -109,7 +162,7 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int
indexError("failed to open file, error %d", errno);
goto END;
}
} else if (ctx->type == TMemory) {
} else if (ctx->type == TMEMORY) {
ctx->mem.buf = taosMemoryCalloc(1, sizeof(char) * capacity);
ctx->mem.cap = capacity;
}
@ -124,14 +177,14 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int
return ctx;
END:
if (ctx->type == TMemory) {
if (ctx->type == TMEMORY) {
taosMemoryFree(ctx->mem.buf);
}
taosMemoryFree(ctx);
return NULL;
}
void idxFileCtxDestroy(IFileCtx* ctx, bool remove) {
if (ctx->type == TMemory) {
if (ctx->type == TMEMORY) {
taosMemoryFree(ctx->mem.buf);
} else {
ctx->flush(ctx);
@ -183,6 +236,7 @@ int idxFileWrite(IdxFstFile* write, uint8_t* buf, uint32_t len) {
write->summer = taosCalcChecksum(write->summer, buf, len);
return len;
}
int idxFileRead(IdxFstFile* write, uint8_t* buf, uint32_t len) {
if (write == NULL) {
return 0;

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

@ -90,7 +90,7 @@ static int32_t (*tfSearch[][QUERY_MAX])(void* reader, SIndexTerm* tem, SIdxTRslt
{tfSearchEqual_JSON, tfSearchPrefix_JSON, tfSearchSuffix_JSON, tfSearchRegex_JSON, tfSearchLessThan_JSON,
tfSearchLessEqual_JSON, tfSearchGreaterThan_JSON, tfSearchGreaterEqual_JSON, tfSearchRange_JSON}};
TFileCache* tfileCacheCreate(const char* path) {
TFileCache* tfileCacheCreate(SIndex* idx, const char* path) {
TFileCache* tcache = taosMemoryCalloc(1, sizeof(TFileCache));
if (tcache == NULL) {
return NULL;
@ -103,17 +103,20 @@ TFileCache* tfileCacheCreate(const char* path) {
for (size_t i = 0; i < taosArrayGetSize(files); i++) {
char* file = taosArrayGetP(files, i);
IFileCtx* wc = idxFileCtxCreate(TFile, file, true, 1024 * 1024 * 64);
if (wc == NULL) {
IFileCtx* ctx = idxFileCtxCreate(TFILE, file, true, 1024 * 1024 * 64);
if (ctx == NULL) {
indexError("failed to open index:%s", file);
goto End;
}
ctx->lru = idx->lru;
TFileReader* reader = tfileReaderCreate(wc);
TFileReader* reader = tfileReaderCreate(ctx);
if (reader == NULL) {
indexInfo("skip invalid file: %s", file);
continue;
}
reader->lru = idx->lru;
TFileHeader* header = &reader->header;
ICacheKey key = {.suid = header->suid, .colName = header->colName, .nColName = (int32_t)strlen(header->colName)};
@ -160,9 +163,8 @@ TFileReader* tfileCacheGet(TFileCache* tcache, ICacheKey* key) {
return *reader;
}
void tfileCachePut(TFileCache* tcache, ICacheKey* key, TFileReader* reader) {
char buf[128] = {0};
int32_t sz = idxSerialCacheKey(key, buf);
// remove last version index reader
char buf[128] = {0};
int32_t sz = idxSerialCacheKey(key, buf);
TFileReader** p = taosHashGet(tcache->tableCache, buf, sz);
if (p != NULL && *p != NULL) {
TFileReader* oldRdr = *p;
@ -493,7 +495,7 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int64_t version, const c
char fullname[256] = {0};
tfileGenFileFullName(fullname, path, suid, colName, version);
// indexInfo("open write file name %s", fullname);
IFileCtx* wcx = idxFileCtxCreate(TFile, fullname, false, 1024 * 1024 * 64);
IFileCtx* wcx = idxFileCtxCreate(TFILE, fullname, false, 1024 * 1024 * 64);
if (wcx == NULL) {
return NULL;
}
@ -506,16 +508,17 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int64_t version, const c
return tfileWriterCreate(wcx, &tfh);
}
TFileReader* tfileReaderOpen(char* path, uint64_t suid, int64_t version, const char* colName) {
TFileReader* tfileReaderOpen(SIndex* idx, uint64_t suid, int64_t version, const char* colName) {
char fullname[256] = {0};
tfileGenFileFullName(fullname, path, suid, colName, version);
tfileGenFileFullName(fullname, idx->path, suid, colName, version);
IFileCtx* wc = idxFileCtxCreate(TFile, fullname, true, 1024 * 1024 * 1024);
IFileCtx* wc = idxFileCtxCreate(TFILE, fullname, true, 1024 * 1024 * 1024);
if (wc == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
indexError("failed to open readonly file: %s, reason: %s", fullname, terrstr());
return NULL;
}
wc->lru = idx->lru;
indexTrace("open read file name:%s, file size: %" PRId64 "", wc->file.buf, wc->file.size);
TFileReader* reader = tfileReaderCreate(wc);
@ -598,17 +601,11 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
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));
// indexInfo("tfile write data size: %d", tw->ctx->size(tw->ctx));
indexInfo("success to write data: %s, offset: %d len: %d", v->colVal, v->offset,
(int)taosArrayGetSize(v->tableId));
}
}
fstBuilderFinish(tw->fb);
fstBuilderDestroy(tw->fb);
tw->fb = NULL;
tfileWriteFooter(tw);
return 0;
}
@ -627,8 +624,8 @@ void tfileWriterDestroy(TFileWriter* tw) {
taosMemoryFree(tw);
}
IndexTFile* idxTFileCreate(const char* path) {
TFileCache* cache = tfileCacheCreate(path);
IndexTFile* idxTFileCreate(SIndex* idx, const char* path) {
TFileCache* cache = tfileCacheCreate(idx, path);
if (cache == NULL) {
return NULL;
}
@ -859,18 +856,6 @@ static int tfileWriteData(TFileWriter* write, TFileValue* tval) {
return 0;
}
return -1;
// if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) {
// FstSlice key = fstSliceCreate((uint8_t*)(tval->colVal), (size_t)strlen(tval->colVal));
// if (fstBuilderInsert(write->fb, key, tval->offset)) {
// fstSliceDestroy(&key);
// return 0;
// }
// fstSliceDestroy(&key);
// return -1;
//} else {
// // handle other type later
//}
}
static int tfileWriteFooter(TFileWriter* write) {
char buf[sizeof(FILE_MAGIC_NUMBER) + 1] = {0};
@ -887,6 +872,7 @@ static int tfileReaderLoadHeader(TFileReader* reader) {
char buf[TFILE_HEADER_SIZE] = {0};
int64_t nread = reader->ctx->readFrom(reader->ctx, buf, sizeof(buf), 0);
if (nread == -1) {
indexError("actual Read: %d, to read: %d, errno: %d, filename: %s", (int)(nread), (int)sizeof(buf), errno,
reader->ctx->file.buf);
@ -914,7 +900,7 @@ static int tfileReaderLoadFst(TFileReader* reader) {
int64_t cost = taosGetTimestampUs() - ts;
indexInfo("nread = %d, and fst offset=%d, fst size: %d, filename: %s, file size: %" PRId64 ", time cost: %" PRId64
"us",
nread, reader->header.fstOffset, fstSize, ctx->file.buf, ctx->file.size, cost);
nread, reader->header.fstOffset, fstSize, ctx->file.buf, size, cost);
// we assuse fst size less than FST_MAX_SIZE
assert(nread > 0 && nread <= fstSize);

12
source/libs/index/test/fstTest.cc
View File

@ -19,7 +19,7 @@ class FstWriter {
public:
FstWriter() {
taosRemoveFile(fileName.c_str());
_wc = idxFileCtxCreate(TFile, fileName.c_str(), false, 64 * 1024 * 1024);
_wc = idxFileCtxCreate(TFILE, fileName.c_str(), false, 64 * 1024 * 1024);
_b = fstBuilderCreate(_wc, 0);
}
bool Put(const std::string& key, uint64_t val) {
@ -34,7 +34,7 @@ class FstWriter {
return ok;
}
~FstWriter() {
fstBuilderFinish(_b);
// fstBuilderFinish(_b);
fstBuilderDestroy(_b);
idxFileCtxDestroy(_wc, false);
@ -48,7 +48,7 @@ class FstWriter {
class FstReadMemory {
public:
FstReadMemory(int32_t size, const std::string& fileName = TD_TMP_DIR_PATH "tindex.tindex") {
_wc = idxFileCtxCreate(TFile, fileName.c_str(), true, 64 * 1024);
_wc = idxFileCtxCreate(TFILE, fileName.c_str(), true, 64 * 1024);
_w = idxFileCreate(_wc);
_size = size;
memset((void*)&_s, 0, sizeof(_s));
@ -598,7 +598,9 @@ void fst_get(Fst* fst) {
void validateTFile(char* arg) {
std::thread threads[NUM_OF_THREAD];
// std::vector<std::thread> threads;
TFileReader* reader = tfileReaderOpen(arg, 0, 20000000, "tag1");
SIndex* index = (SIndex*)taosMemoryCalloc(1, sizeof(SIndex));
index->path = strdup(arg);
TFileReader* reader = tfileReaderOpen(index, 0, 20000000, "tag1");
for (int i = 0; i < NUM_OF_THREAD; i++) {
threads[i] = std::thread(fst_get, reader->fst);
@ -617,7 +619,7 @@ void iterTFileReader(char* path, char* uid, char* colName, char* ver) {
uint64_t suid = atoi(uid);
int version = atoi(ver);
TFileReader* reader = tfileReaderOpen(path, suid, version, colName);
TFileReader* reader = tfileReaderOpen(NULL, suid, version, colName);
Iterate* iter = tfileIteratorCreate(reader);
bool tn = iter ? iter->next(iter) : false;

5
source/libs/index/test/fstUT.cc
View File

@ -39,7 +39,7 @@ static void EnvCleanup() {}
class FstWriter {
public:
FstWriter() {
_wc = idxFileCtxCreate(TFile, tindex, false, 64 * 1024 * 1024);
_wc = idxFileCtxCreate(TFILE, tindex, false, 64 * 1024 * 1024);
_b = fstBuilderCreate(_wc, 0);
}
bool Put(const std::string& key, uint64_t val) {
@ -54,7 +54,6 @@ class FstWriter {
return ok;
}
~FstWriter() {
fstBuilderFinish(_b);
fstBuilderDestroy(_b);
idxFileCtxDestroy(_wc, false);
@ -68,7 +67,7 @@ class FstWriter {
class FstReadMemory {
public:
FstReadMemory(size_t size) {
_wc = idxFileCtxCreate(TFile, tindex, true, 64 * 1024);
_wc = idxFileCtxCreate(TFILE, tindex, true, 64 * 1024);
_w = idxFileCreate(_wc);
_size = size;
memset((void*)&_s, 0, sizeof(_s));

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

@ -50,7 +50,7 @@ class DebugInfo {
class FstWriter {
public:
FstWriter() {
_wc = idxFileCtxCreate(TFile, TD_TMP_DIR_PATH "tindex", false, 64 * 1024 * 1024);
_wc = idxFileCtxCreate(TFILE, TD_TMP_DIR_PATH "tindex", false, 64 * 1024 * 1024);
_b = fstBuilderCreate(NULL, 0);
}
bool Put(const std::string& key, uint64_t val) {
@ -60,7 +60,7 @@ class FstWriter {
return ok;
}
~FstWriter() {
fstBuilderFinish(_b);
// fstBuilderFinish(_b);
fstBuilderDestroy(_b);
idxFileCtxDestroy(_wc, false);
@ -74,7 +74,7 @@ class FstWriter {
class FstReadMemory {
public:
FstReadMemory(size_t size) {
_wc = idxFileCtxCreate(TFile, TD_TMP_DIR_PATH "tindex", true, 64 * 1024);
_wc = idxFileCtxCreate(TFILE, TD_TMP_DIR_PATH "tindex", true, 64 * 1024);
_w = idxFileCreate(_wc);
_size = size;
memset((void*)&_s, 0, sizeof(_s));
@ -292,7 +292,7 @@ class IndexEnv : public ::testing::Test {
virtual void SetUp() {
initLog();
taosRemoveDir(path);
opts = indexOptsCreate();
opts = indexOptsCreate(1024 * 1024 * 8);
int ret = indexOpen(opts, path, &index);
assert(ret == 0);
}
@ -391,13 +391,15 @@ class TFileObj {
fileName_ = path;
IFileCtx* ctx = idxFileCtxCreate(TFile, path.c_str(), false, 64 * 1024 * 1024);
IFileCtx* ctx = idxFileCtxCreate(TFILE, path.c_str(), false, 64 * 1024 * 1024);
ctx->lru = taosLRUCacheInit(1024 * 1024 * 4, -1, .5);
writer_ = tfileWriterCreate(ctx, &header);
return writer_ != NULL ? true : false;
}
bool InitReader() {
IFileCtx* ctx = idxFileCtxCreate(TFile, fileName_.c_str(), true, 64 * 1024 * 1024);
IFileCtx* ctx = idxFileCtxCreate(TFILE, fileName_.c_str(), true, 64 * 1024 * 1024);
ctx->lru = taosLRUCacheInit(1024 * 1024 * 4, -1, .5);
reader_ = tfileReaderCreate(ctx);
return reader_ != NULL ? true : false;
}
@ -657,7 +659,7 @@ TEST_F(IndexCacheEnv, cache_test) {
{
std::string colVal("v3");
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexTermQuery query = {term, QUERY_TERM};
SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid));
STermValueType valType;
@ -672,7 +674,7 @@ TEST_F(IndexCacheEnv, cache_test) {
{
std::string colVal("v2");
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexTermQuery query = {term, QUERY_TERM};
SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid));
STermValueType valType;
@ -698,7 +700,8 @@ class IndexObj {
taosMkDir(dir.c_str());
}
taosMkDir(dir.c_str());
int ret = indexOpen(&opts, dir.c_str(), &idx);
opts = indexOptsCreate(1024 * 1024 * 4);
int ret = indexOpen(opts, dir.c_str(), &idx);
if (ret != 0) {
// opt
std::cout << "failed to open index: %s" << dir << std::endl;
@ -707,7 +710,7 @@ class IndexObj {
}
void Del(const std::string& colName, const std::string& colVal, uint64_t uid) {
SIndexTerm* term = indexTermCreateT(0, DEL_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
Put(terms, uid);
@ -716,7 +719,7 @@ class IndexObj {
int WriteMillonData(const std::string& colName, const std::string& colVal = "Hello world",
size_t numOfTable = 100 * 10000) {
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = 0; i < numOfTable; i++) {
@ -738,7 +741,7 @@ class IndexObj {
tColVal[taosRand() % colValSize] = 'a' + k % 26;
}
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
tColVal.c_str(), tColVal.size());
tColVal.c_str(), tColVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t j = 0; j < skip; j++) {
@ -774,7 +777,7 @@ class IndexObj {
int SearchOne(const std::string& colName, const std::string& colVal) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
indexMultiTermQueryAdd(mq, term, QUERY_TERM);
SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t));
@ -796,7 +799,7 @@ class IndexObj {
int SearchOneTarget(const std::string& colName, const std::string& colVal, uint64_t val) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
indexMultiTermQueryAdd(mq, term, QUERY_TERM);
SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t));
@ -821,7 +824,7 @@ class IndexObj {
void PutOne(const std::string& colName, const std::string& colVal) {
SIndexMultiTerm* terms = indexMultiTermCreate();
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
Put(terms, 10);
indexMultiTermDestroy(terms);
@ -829,7 +832,7 @@ class IndexObj {
void PutOneTarge(const std::string& colName, const std::string& colVal, uint64_t val) {
SIndexMultiTerm* terms = indexMultiTermCreate();
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
Put(terms, val);
indexMultiTermDestroy(terms);
@ -845,10 +848,10 @@ class IndexObj {
}
private:
SIndexOpts opts;
SIndex* idx;
int numOfWrite;
int numOfRead;
SIndexOpts* opts;
SIndex* idx;
int numOfWrite;
int numOfRead;
};
class IndexEnv2 : public ::testing::Test {
@ -875,7 +878,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
std::string colName("tag1"), colVal("Hello");
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = 0; i < targetSize; i++) {
@ -890,7 +893,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
std::string colName("tag1"), colVal("hello");
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = 0; i < size; i++) {
@ -905,7 +908,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
std::string colName("tag1"), colVal("Hello");
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = size * 3; i < size * 4; i++) {
@ -920,7 +923,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
std::string colName("tag1"), colVal("Hello");
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
indexMultiTermQueryAdd(mq, term, QUERY_TERM);
SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t));
@ -943,7 +946,7 @@ TEST_F(IndexEnv2, testEmptyIndexOpen) {
std::string colName("tag1"), colVal("Hello");
SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = 0; i < targetSize; i++) {

41
source/libs/index/test/jsonUT.cc
View File

@ -54,13 +54,12 @@ class JsonEnv : public ::testing::Test {
printf("set up\n");
initLog();
opts = indexOptsCreate();
opts = indexOptsCreate(1024 * 1024 * 4);
int ret = indexJsonOpen(opts, dir.c_str(), &index);
assert(ret == 0);
}
virtual void TearDown() {
indexJsonClose(index);
indexOptsDestroy(opts);
printf("destory\n");
taosMsleep(1000);
}
@ -71,7 +70,7 @@ class JsonEnv : public ::testing::Test {
static void WriteData(SIndexJson* index, const std::string& colName, int8_t dtype, void* data, int dlen, int tableId,
int8_t operType = ADD_VALUE) {
SIndexTerm* term = indexTermCreateT(1, (SIndexOperOnColumn)operType, dtype, colName.c_str(), colName.size(),
(const char*)data, dlen);
(const char*)data, dlen);
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
indexJsonPut(index, terms, (int64_t)tableId);
@ -82,7 +81,7 @@ static void WriteData(SIndexJson* index, const std::string& colName, int8_t dtyp
static void delData(SIndexJson* index, const std::string& colName, int8_t dtype, void* data, int dlen, int tableId,
int8_t operType = DEL_VALUE) {
SIndexTerm* term = indexTermCreateT(1, (SIndexOperOnColumn)operType, dtype, colName.c_str(), colName.size(),
(const char*)data, dlen);
(const char*)data, dlen);
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
indexJsonPut(index, terms, (int64_t)tableId);
@ -108,7 +107,7 @@ TEST_F(JsonEnv, testWrite) {
std::string colVal("ab");
for (int i = 0; i < 100; i++) {
SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
indexJsonPut(index, terms, i);
@ -147,7 +146,7 @@ TEST_F(JsonEnv, testWrite) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_TERM);
@ -205,7 +204,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_TERM);
@ -220,7 +219,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN);
@ -235,7 +234,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
colVal.c_str(), colVal.size());
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL);
@ -305,7 +304,7 @@ TEST_F(JsonEnv, testWriteJsonNumberData) {
int val = 15;
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_TERM);
@ -319,7 +318,7 @@ TEST_F(JsonEnv, testWriteJsonNumberData) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN);
@ -334,7 +333,7 @@ TEST_F(JsonEnv, testWriteJsonNumberData) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(int));
(const char*)&val, sizeof(int));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL);
@ -349,7 +348,7 @@ TEST_F(JsonEnv, testWriteJsonNumberData) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_LESS_THAN);
@ -364,7 +363,7 @@ TEST_F(JsonEnv, testWriteJsonNumberData) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_LESS_EQUAL);
@ -407,7 +406,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_TERM);
@ -421,7 +420,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(int));
(const char*)&val, sizeof(int));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN);
@ -436,7 +435,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL);
@ -450,7 +449,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN);
@ -464,7 +463,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_LESS_EQUAL);
@ -493,7 +492,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_LESS_THAN);
@ -521,7 +520,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(),
(const char*)&val, sizeof(val));
(const char*)&val, sizeof(val));
SArray* result = taosArrayInit(1, sizeof(uint64_t));
indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL);