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

Merge pull request #9392 from taosdata/feature/index_cache 

Feature/index cache
This commit is contained in:
Shengliang Guan 2021-12-26 10:58:15 +08:00 committed by GitHub
parent 60be4f7331 0365c67b81
commit 3052e8e154
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 310 additions and 198 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
source/libs/index/inc/index_cache.h
View File

@ -37,6 +37,17 @@ typedef struct IndexCache {
SSkipList* skiplist;
} IndexCache;
typedef struct CacheTerm {
// key
int32_t colId;
int32_t nColVal;
char* colVal;
int32_t version;
// value
uint64_t uid;
int8_t colType;
SIndexOperOnColumn operaType;
} CacheTerm;
//
IndexCache* indexCacheCreate();
@ -47,6 +58,7 @@ int indexCachePut(void* cache, SIndexTerm* term, int16_t colId, int32_t version,
// int indexCacheGet(void *cache, uint64_t *rst);
int indexCacheSearch(void* cache, SIndexTermQuery* query, int16_t colId, int32_t version, SArray* result, STermValueType* s);
void indexCacheDebug(IndexCache* cache);
#ifdef __cplusplus
}
#endif

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

@ -41,7 +41,7 @@ static pthread_once_t isInit = PTHREAD_ONCE_INIT;
static void indexInit();
static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* term, SArray** result);
static int indexFlushCacheToTindex(SIndex* sIdx);
static int indexFlushCacheTFile(SIndex* sIdx);
static void indexInterResultsDestroy(SArray* results);
static int indexMergeFinalResults(SArray* interResults, EIndexOperatorType oType, SArray* finalResult);
@ -353,7 +353,7 @@ static int indexMergeFinalResults(SArray* interResults, EIndexOperatorType oType
}
return 0;
}
static int indexFlushCacheToTindex(SIndex* sIdx) {
static int indexFlushCacheTFile(SIndex* sIdx) {
if (sIdx == NULL) { return -1; }
indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid);

201
source/libs/index/src/index_cache.c
View File

@ -20,81 +20,88 @@
#define MAX_INDEX_KEY_LEN 256 // test only, change later
// 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) + sizeof(p->operType))
//#define CACHE_KEY_LEN(p) \
// (sizeof(int32_t) + sizeof(uint16_t) + sizeof(p->colType) + sizeof(p->nColVal) + p->nColVal + sizeof(uint64_t) + sizeof(p->operType))
static char* getIndexKey(const void* pData) {
return NULL;
static void cacheTermDestroy(CacheTerm* ct) {
if (ct == NULL) { return; }
free(ct->colVal);
free(ct);
}
static char* getIndexKey(const void* pData) {
CacheTerm* p = (CacheTerm*)pData;
return (char*)p;
}
static int32_t compareKey(const void* l, const void* r) {
char* lp = (char*)l;
char* rp = (char*)r;
CacheTerm* lt = (CacheTerm*)l;
CacheTerm* rt = (CacheTerm*)r;
// skip total len, not compare
int32_t ll, rl; // len
memcpy(&ll, lp, sizeof(int32_t));
memcpy(&rl, rp, sizeof(int32_t));
lp += sizeof(int32_t);
rp += sizeof(int32_t);
// compare colId
if (lt->colId != rt->colId) { return lt->colId - rt->colId; }
// compare field id
int16_t lf, rf; // field id
memcpy(&lf, lp, sizeof(lf));
memcpy(&rf, rp, sizeof(rf));
if (lf != rf) { return lf < rf ? -1 : 1; }
lp += sizeof(lf);
rp += sizeof(rf);
// compare field type
int8_t lft, rft;
memcpy(&lft, lp, sizeof(lft));
memcpy(&rft, rp, sizeof(rft));
lp += sizeof(lft);
rp += sizeof(rft);
assert(rft == rft);
// skip value len
int32_t lfl, rfl;
memcpy(&lfl, lp, sizeof(lfl));
memcpy(&rfl, rp, sizeof(rfl));
lp += sizeof(lfl);
rp += sizeof(rfl);
// compare value
int32_t i, j;
for (i = 0, j = 0; i < lfl && j < rfl; i++, j++) {
if (lp[i] == rp[j]) {
// compare colVal
int i, j;
for (i = 0, j = 0; i < lt->nColVal && j < rt->nColVal; i++, j++) {
if (lt->colVal[i] == rt->colVal[j]) {
continue;
} else {
return lp[i] < rp[j] ? -1 : 1;
return lt->colVal[i] < rt->colVal[j] ? -1 : 1;
}
}
if (i < lfl) {
if (i < lt->nColVal) {
return 1;
} else if (j < rfl) {
} else if (j < rt->nColVal) {
return -1;
}
lp += lfl;
rp += rfl;
// compare version
// skip uid
uint64_t lu, ru;
memcpy(&lu, lp, sizeof(lu));
memcpy(&ru, rp, sizeof(ru));
lp += sizeof(lu);
rp += sizeof(ru);
return rt->version - lt->version;
// compare version, desc order
int32_t lv, rv;
memcpy(&lv, lp, sizeof(lv));
memcpy(&rv, rp, sizeof(rv));
if (lv != rv) { return lv > rv ? -1 : 1; }
// char* lp = (char*)l;
// char* rp = (char*)r;
lp += sizeof(lv);
rp += sizeof(rv);
// not care item type
//// compare col id
// int16_t lf, rf; // cold id
// memcpy(&lf, lp, sizeof(lf));
// memcpy(&rf, rp, sizeof(rf));
// if (lf != rf) { return lf < rf ? -1 : 1; }
return 0;
// lp += sizeof(lf);
// rp += sizeof(rf);
//// skip value len
// int32_t lfl, rfl;
// memcpy(&lfl, lp, sizeof(lfl));
// memcpy(&rfl, rp, sizeof(rfl));
// lp += sizeof(lfl);
// rp += sizeof(rfl);
//// compare value
// int32_t i, j;
// for (i = 0, j = 0; i < lfl && j < rfl; i++, j++) {
// if (lp[i] == rp[j]) {
// continue;
// } else {
// return lp[i] < rp[j] ? -1 : 1;
// }
//}
// if (i < lfl) {
// return 1;
//} else if (j < rfl) {
// return -1;
//}
// lp += lfl;
// rp += rfl;
//// compare version, desc order
// int32_t lv, rv;
// memcpy(&lv, lp, sizeof(lv));
// memcpy(&rv, rp, sizeof(rv));
// if (lv != rv) { return lv < rv ? 1 : -1; }
// return 0;
}
IndexCache* indexCacheCreate() {
IndexCache* cache = calloc(1, sizeof(IndexCache));
@ -106,6 +113,18 @@ IndexCache* indexCacheCreate() {
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, MAX_INDEX_KEY_LEN, compareKey, SL_ALLOW_DUP_KEY, getIndexKey);
return cache;
}
void indexCacheDebug(IndexCache* cache) {
SSkipListIterator* iter = tSkipListCreateIter(cache->skiplist);
while (tSkipListIterNext(iter)) {
SSkipListNode* node = tSkipListIterGet(iter);
CacheTerm* ct = (CacheTerm*)SL_GET_NODE_DATA(node);
if (ct != NULL) {
// TODO, add more debug info
indexInfo("{colId:%d, colVal: %s, version: %d} \t", ct->colId, ct->colVal, ct->version);
}
}
tSkipListDestroyIter(iter);
}
void indexCacheDestroy(void* cache) {
IndexCache* pCache = cache;
@ -119,24 +138,20 @@ int indexCachePut(void* cache, SIndexTerm* term, int16_t colId, int32_t version,
IndexCache* pCache = cache;
// encode data
int32_t total = CACHE_KEY_LEN(term);
CacheTerm* ct = calloc(1, sizeof(CacheTerm));
if (cache == NULL) { return -1; }
// set up key
ct->colId = colId;
ct->colType = term->colType;
ct->nColVal = term->nColVal;
ct->colVal = (char*)calloc(1, sizeof(char) * (ct->nColVal + 1));
memcpy(ct->colVal, term->colVal, ct->nColVal);
ct->version = version;
char* buf = calloc(1, total);
char* p = buf;
ct->uid = uid;
ct->operaType = term->operType;
SERIALIZE_VAR_TO_BUF(p, total, int32_t);
SERIALIZE_VAR_TO_BUF(p, colId, int16_t);
SERIALIZE_MEM_TO_BUF(p, term, colType);
SERIALIZE_MEM_TO_BUF(p, term, nColVal);
SERIALIZE_STR_MEM_TO_BUF(p, term, colVal, term->nColVal);
SERIALIZE_VAR_TO_BUF(p, version, int32_t);
SERIALIZE_VAR_TO_BUF(p, uid, uint64_t);
SERIALIZE_MEM_TO_BUF(p, term, operType);
tSkipListPut(pCache->skiplist, (void*)buf);
tSkipListPut(pCache->skiplist, (char*)ct);
return 0;
// encode end
}
@ -150,8 +165,39 @@ int indexCacheSearch(void* cache, SIndexTermQuery* query, int16_t colId, int32_t
SIndexTerm* term = query->term;
EIndexQueryType qtype = query->qType;
int32_t keyLen = CACHE_KEY_LEN(term);
char* buf = calloc(1, keyLen);
CacheTerm* ct = calloc(1, sizeof(CacheTerm));
if (ct == NULL) { return -1; }
ct->colId = colId;
ct->nColVal = term->nColVal;
ct->colVal = calloc(1, sizeof(char) * (ct->nColVal + 1));
memcpy(ct->colVal, term->colVal, ct->nColVal);
ct->version = version;
char* key = getIndexKey(ct);
// TODO handle multi situation later, and refactor
SSkipListIterator* iter = tSkipListCreateIterFromVal(pCache->skiplist, key, TSDB_DATA_TYPE_BINARY, TSDB_ORDER_ASC);
while (tSkipListIterNext(iter)) {
SSkipListNode* node = tSkipListIterGet(iter);
if (node != NULL) {
CacheTerm* c = (CacheTerm*)SL_GET_NODE_DATA(node);
if (c->operaType == ADD_VALUE || qtype == QUERY_TERM) {
if (c->nColVal == ct->nColVal && strncmp(c->colVal, ct->colVal, c->nColVal) == 0) {
taosArrayPush(result, &c->uid);
*s = kTypeValue;
} else {
break;
}
} else if (c->operaType == DEL_VALUE) {
// table is del, not need
*s = kTypeDeletion;
break;
}
}
}
tSkipListDestroyIter(iter);
cacheTermDestroy(ct);
// int32_t keyLen = CACHE_KEY_LEN(term);
// char* buf = calloc(1, keyLen);
if (qtype == QUERY_TERM) {
//
} else if (qtype == QUERY_PREFIX) {
@ -161,6 +207,5 @@ int indexCacheSearch(void* cache, SIndexTermQuery* query, int16_t colId, int32_t
} else if (qtype == QUERY_REGEX) {
//
}
return 0;
}

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

@ -462,7 +462,9 @@ static int tfileCompare(const void* a, const void* b) {
size_t aLen = strlen(aName);
size_t bLen = strlen(bName);
return strncmp(aName, bName, aLen > bLen ? aLen : bLen);
int ret = strncmp(aName, bName, aLen > bLen ? aLen : bLen);
if (ret == 0) { return ret; }
return ret < 0 ? -1 : 1;
}
// tfile name suid-colId-version.tindex
static void tfileGenFileName(char* filename, uint64_t suid, int colId, int version) {

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

@ -16,12 +16,30 @@
#include <string>
#include "index.h"
#include "indexInt.h"
#include "index_cache.h"
#include "index_fst.h"
#include "index_fst_counting_writer.h"
#include "index_fst_util.h"
#include "index_tfile.h"
#include "tskiplist.h"
#include "tutil.h"
using namespace std;
class DebugInfo {
public:
DebugInfo(const char* str) : info(str) {
std::cout << "------------" << info << "\t"
<< "begin"
<< "-------------" << std::endl;
}
~DebugInfo() {
std::cout << "-----------" << info << "\t"
<< "end"
<< "--------------" << std::endl;
}
private:
std::string info;
};
class FstWriter {
public:
FstWriter() {
@ -110,58 +128,6 @@ class FstReadMemory {
size_t _size;
};
// TEST(IndexTest, index_create_test) {
// SIndexOpts *opts = indexOptsCreate();
// SIndex *index = indexOpen(opts, "./test");
// if (index == NULL) {
// std::cout << "index open failed" << std::endl;
// }
//
//
// // write
// for (int i = 0; i < 100000; i++) {
// SIndexMultiTerm* terms = indexMultiTermCreate();
// std::string val = "field";
//
// indexMultiTermAdd(terms, "tag1", strlen("tag1"), val.c_str(), val.size());
//
// val.append(std::to_string(i));
// indexMultiTermAdd(terms, "tag2", strlen("tag2"), val.c_str(), val.size());
//
// val.insert(0, std::to_string(i));
// indexMultiTermAdd(terms, "tag3", strlen("tag3"), val.c_str(), val.size());
//
// val.append("const");
// indexMultiTermAdd(terms, "tag4", strlen("tag4"), val.c_str(), val.size());
//
//
// indexPut(index, terms, i);
// indexMultiTermDestroy(terms);
// }
//
//
// // query
// SIndexMultiTermQuery *multiQuery = indexMultiTermQueryCreate(MUST);
//
// indexMultiTermQueryAdd(multiQuery, "tag1", strlen("tag1"), "field", strlen("field"), QUERY_PREFIX);
// indexMultiTermQueryAdd(multiQuery, "tag3", strlen("tag3"), "0field0", strlen("0field0"), QUERY_TERM);
//
// SArray *result = (SArray *)taosArrayInit(10, sizeof(int));
// indexSearch(index, multiQuery, result);
//
// std::cout << "taos'size : " << taosArrayGetSize(result) << std::endl;
// for (int i = 0; i < taosArrayGetSize(result); i++) {
// int *v = (int *)taosArrayGet(result, i);
// std::cout << "value --->" << *v << std::endl;
// }
// // add more test case
// indexMultiTermQueryDestroy(multiQuery);
//
// indexOptsDestroy(opts);
// indexClose(index);
// //
//}
#define L 100
#define M 100
#define N 100
@ -183,7 +149,6 @@ 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++) {
@ -218,7 +183,6 @@ void checkFstPerf() {
Performance_fstReadRecords(m);
delete m;
}
void checkFstPrefixSearch() {
FstWriter* fw = new FstWriter;
int64_t s = taosGetTimestampUs();
@ -296,7 +260,6 @@ void validateFst() {
}
delete m;
}
class IndexEnv : public ::testing::Test {
protected:
virtual void SetUp() {
@ -315,44 +278,51 @@ class IndexEnv : public ::testing::Test {
SIndex* index;
};
// TEST_F(IndexEnv, testPut) {
// // single index column
// {
// std::string colName("tag1"), colVal("Hello world");
// 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);
//
// for (size_t i = 0; i < 100; i++) {
// int tableId = i;
// int ret = indexPut(index, terms, tableId);
// assert(ret == 0);
// }
// indexMultiTermDestroy(terms);
// }
// // multi index column
// {
/// TEST_F(IndexEnv, testPut) {
// / // single index column
// / {
// / std::string colName("tag1"), colVal("Hello world");
// / SIndexTerm* term =
// indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), / colVal.size());
// SIndexMultiTerm* terms = indexMultiTermCreate();
// {
// std::string colName("tag1"), colVal("Hello world");
// SIndexTerm* term =
// indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
// indexMultiTermAdd(terms, term);
// / / for (size_t i = 0; i < 100; i++) {
// / int tableId = i;
// / int ret = indexPut(index, terms, tableId);
// / assert(ret == 0);
// /
// }
// {
// std::string colName("tag2"), colVal("Hello world");
// SIndexTerm* term =
// indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
// indexMultiTermAdd(terms, term);
// / indexMultiTermDestroy(terms);
// /
// }
//
// for (int i = 0; i < 100; i++) {
// int tableId = i;
// int ret = indexPut(index, terms, tableId);
// assert(ret == 0);
// / // multi index column
// / {
// / SIndexMultiTerm* terms = indexMultiTermCreate();
// / {
// / std::string colName("tag1"), colVal("Hello world");
// / SIndexTerm* term =
// / indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
// / indexMultiTermAdd(terms, term);
// /
// }
// indexMultiTermDestroy(terms);
// / {
// / std::string colName("tag2"), colVal("Hello world");
// / SIndexTerm* term =
// / indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
// / indexMultiTermAdd(terms, term);
// /
// }
// //
// / / for (int i = 0; i < 100; i++) {
// / int tableId = i;
// / int ret = indexPut(index, terms, tableId);
// / assert(ret == 0);
// /
// }
// / indexMultiTermDestroy(terms);
// /
// }
// / //
// /
//}
class TFileObj {
@ -421,6 +391,7 @@ class TFileObj {
int colId_;
};
class IndexTFileEnv : public ::testing::Test {
protected:
virtual void SetUp() {
@ -428,22 +399,6 @@ class IndexTFileEnv : public ::testing::Test {
taosMkDir(dir.c_str());
tfInit();
fObj = new TFileObj(dir, colName);
// std::string colName("voltage");
// header.suid = 1;
// header.version = 1;
// memcpy(header.colName, colName.c_str(), colName.size());
// header.colType = TSDB_DATA_TYPE_BINARY;
// std::string path(dir);
// int colId = 2;
// char buf[64] = {0};
// sprintf(buf, "%" PRIu64 "-%d-%d.tindex", header.suid, colId, header.version);
// path.append("/").append(buf);
// ctx = writerCtxCreate(TFile, path.c_str(), false, 64 * 1024 * 1024);
// twrite = tfileWriterCreate(ctx, &header);
}
virtual void TearDown() {
@ -460,16 +415,8 @@ class IndexTFileEnv : public ::testing::Test {
int coldId = 2;
int version = 1;
int colType = TSDB_DATA_TYPE_BINARY;
// WriterCtx* ctx = NULL;
// TFileHeader header;
// TFileWriter* twrite = NULL;
};
// static TFileWriter* genTFileWriter(const char* path, TFileHeader* header) {
// char buf[128] = {0};
// WriterCtx* ctx = writerCtxCreate(TFile, path, false, )
//}
static TFileValue* genTFileValue(const char* val) {
TFileValue* tv = (TFileValue*)calloc(1, sizeof(TFileValue));
int32_t vlen = strlen(val) + 1;
@ -489,10 +436,9 @@ static void destroyTFileValue(void* val) {
taosArrayDestroy(tv->tableId);
free(tv);
}
TEST_F(IndexTFileEnv, test_tfile_write) {
TFileValue* v1 = genTFileValue("c");
TFileValue* v2 = genTFileValue("a");
TFileValue* v2 = genTFileValue("ab");
TFileValue* v3 = genTFileValue("b");
TFileValue* v4 = genTFileValue("d");
@ -510,7 +456,7 @@ TEST_F(IndexTFileEnv, test_tfile_write) {
taosArrayDestroy(data);
std::string colName("voltage");
std::string colVal("b");
std::string colVal("ab");
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
SIndexTermQuery query = {.term = term, .qType = QUERY_TERM};
@ -521,3 +467,110 @@ TEST_F(IndexTFileEnv, test_tfile_write) {
// tfileWriterDestroy(twrite);
}
class CacheObj {
public:
CacheObj() {
// TODO
cache = indexCacheCreate();
}
int Put(SIndexTerm* term, int16_t colId, int32_t version, uint64_t uid) {
int ret = indexCachePut(cache, term, colId, version, uid);
if (ret != 0) {
//
std::cout << "failed to put into cache: " << ret << std::endl;
}
return ret;
}
int Get(SIndexTermQuery* query, int16_t colId, int32_t version, SArray* result, STermValueType* s) {
int ret = indexCacheSearch(cache, query, colId, version, result, s);
if (ret != 0) {
//
std::cout << "failed to get from cache:" << ret << std::endl;
}
return ret;
}
~CacheObj() {
// TODO
indexCacheDestroy(cache);
}
private:
IndexCache* cache = NULL;
};
class IndexCacheEnv : public ::testing::Test {
protected:
virtual void SetUp() {
// TODO
coj = new CacheObj();
}
virtual void TearDown() {
delete coj;
// formate
}
CacheObj* coj;
};
#define MAX_TERM_KEY_LEN 128
TEST_F(IndexCacheEnv, cache_test) {
int version = 0;
int16_t colId = 0;
uint64_t suid = 0;
std::string colName("voltage");
{
std::string colVal("v1");
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++);
}
{
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++);
}
{
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++);
}
{
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++);
}
{
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++);
}
{
std::string colVal("v4");
for (size_t i = 0; i < 100; i++) {
colVal[colVal.size() - 1] = 'a' + i;
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++);
}
}
{
std::string colVal("v3");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
SIndexTermQuery query = {.term = term, .qType = QUERY_TERM};
SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid));
STermValueType valType;
coj->Get(&query, colId, 10000, ret, &valType);
assert(taosArrayGetSize(ret) == 3);
}
{
std::string colVal("v2");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
SIndexTermQuery query = {.term = term, .qType = QUERY_TERM};
SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid));
STermValueType valType;
coj->Get(&query, colId, 10000, ret, &valType);
assert(taosArrayGetSize(ret) == 1);
}
}