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

enh: support tag filter

This commit is contained in:
yihaoDeng 2022-05-28 17:30:44 +08:00
parent 6acbe7e777
commit 0b0253f2d1
9 changed files with 183 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -74,7 +74,7 @@ void indexCacheIteratorDestroy(Iterate* iiter);
int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid); int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid);
// int indexCacheGet(void *cache, uint64_t *rst); // int indexCacheGet(void *cache, uint64_t *rst);
int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTempResult* tr, STermValueType* s); int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* tr, STermValueType* s);
void indexCacheRef(IndexCache* cache); void indexCacheRef(IndexCache* cache);
void indexCacheUnRef(IndexCache* cache); void indexCacheUnRef(IndexCache* cache);

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

@ -105,7 +105,7 @@ TFileReader* tfileGetReaderByCol(IndexTFile* tf, uint64_t suid, char* colName);
TFileReader* tfileReaderOpen(char* path, uint64_t suid, int64_t version, const char* colName); TFileReader* tfileReaderOpen(char* path, uint64_t suid, int64_t version, const char* colName);
TFileReader* tfileReaderCreate(WriterCtx* ctx); TFileReader* tfileReaderCreate(WriterCtx* ctx);
void tfileReaderDestroy(TFileReader* reader); void tfileReaderDestroy(TFileReader* reader);
int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTempResult* tr); int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTRslt* tr);
void tfileReaderRef(TFileReader* reader); void tfileReaderRef(TFileReader* reader);
void tfileReaderUnRef(TFileReader* reader); void tfileReaderUnRef(TFileReader* reader);
@ -120,7 +120,7 @@ int tfileWriterFinish(TFileWriter* tw);
IndexTFile* indexTFileCreate(const char* path); IndexTFile* indexTFileCreate(const char* path);
void indexTFileDestroy(IndexTFile* tfile); void indexTFileDestroy(IndexTFile* tfile);
int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid); int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid);
int indexTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTempResult* tr); int indexTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* tr);
Iterate* tfileIteratorCreate(TFileReader* reader); Iterate* tfileIteratorCreate(TFileReader* reader);
void tfileIteratorDestroy(Iterate* iterator); void tfileIteratorDestroy(Iterate* iterator);

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

@ -66,7 +66,7 @@ extern "C" {
* [1, 4, 5] * [1, 4, 5]
* output:[4, 5] * output:[4, 5]
*/ */
void iIntersection(SArray *interResults, SArray *finalResult); void iIntersection(SArray *in, SArray *out);
/* multi sorted result union /* multi sorted result union
* input: [1, 2, 4, 5] * input: [1, 2, 4, 5]
@ -74,7 +74,7 @@ void iIntersection(SArray *interResults, SArray *finalResult);
* [1, 4, 5] * [1, 4, 5]
* output:[1, 2, 3, 4, 5] * output:[1, 2, 3, 4, 5]
*/ */
void iUnion(SArray *interResults, SArray *finalResult); void iUnion(SArray *in, SArray *out);
/* see example /* see example
* total: [1, 2, 4, 5, 7, 8] * total: [1, 2, 4, 5, 7, 8]
@ -92,19 +92,24 @@ typedef struct {
uint64_t data; uint64_t data;
} SIdxVerdata; } SIdxVerdata;
/*
* index temp result
*
*/
typedef struct { typedef struct {
SArray *total; SArray *total;
SArray *added; SArray *add;
SArray *deled; SArray *del;
} SIdxTempResult; } SIdxTRslt;
SIdxTempResult *idxTempResultCreate(); SIdxTRslt *idxTRsltCreate();
void idxTempResultClear(SIdxTempResult *tr); void idxTRsltClear(SIdxTRslt *tr);
void idxTempResultDestroy(SIdxTempResult *tr); void idxTRsltDestroy(SIdxTRslt *tr);
void idxTRsltMergeTo(SIdxTRslt *tr, SArray *out);
void idxTempResultMergeTo(SIdxTempResult *tr, SArray *result);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

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

@ -29,7 +29,7 @@
#include "lucene++/Lucene_c.h" #include "lucene++/Lucene_c.h"
#endif #endif
#define INDEX_NUM_OF_THREADS 1 #define INDEX_NUM_OF_THREADS 5
#define INDEX_QUEUE_SIZE 200 #define INDEX_QUEUE_SIZE 200
#define INDEX_DATA_BOOL_NULL 0x02 #define INDEX_DATA_BOOL_NULL 0x02
@ -85,7 +85,7 @@ static int indexMergeFinalResults(SArray* interResults, EIndexOperatorType oTyp
static int indexGenTFile(SIndex* index, IndexCache* cache, SArray* batch); static int indexGenTFile(SIndex* index, IndexCache* cache, SArray* batch);
// merge cache and tfile by opera type // merge cache and tfile by opera type
static void indexMergeCacheAndTFile(SArray* result, IterateValue* icache, IterateValue* iTfv, SIdxTempResult* helper); static void indexMergeCacheAndTFile(SArray* result, IterateValue* icache, IterateValue* iTfv, SIdxTRslt* helper);
// static int32_t indexSerialTermKey(SIndexTerm* itm, char* buf); // static int32_t indexSerialTermKey(SIndexTerm* itm, char* buf);
// int32_t indexSerialKey(ICacheKey* key, char* buf); // int32_t indexSerialKey(ICacheKey* key, char* buf);
@ -343,7 +343,7 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
SIdxTempResult* tr = idxTempResultCreate(); SIdxTRslt* tr = idxTRsltCreate();
if (0 == indexCacheSearch(cache, query, tr, &s)) { if (0 == indexCacheSearch(cache, query, tr, &s)) {
if (s == kTypeDeletion) { if (s == kTypeDeletion) {
indexInfo("col: %s already drop by", term->colName); indexInfo("col: %s already drop by", term->colName);
@ -365,12 +365,12 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result
int64_t cost = taosGetTimestampUs() - st; int64_t cost = taosGetTimestampUs() - st;
indexInfo("search cost: %" PRIu64 "us", cost); indexInfo("search cost: %" PRIu64 "us", cost);
idxTempResultMergeTo(tr, *result); idxTRsltMergeTo(tr, *result);
idxTempResultDestroy(tr); idxTRsltDestroy(tr);
return 0; return 0;
END: END:
idxTempResultDestroy(tr); idxTRsltDestroy(tr);
return -1; return -1;
} }
static void indexInterResultsDestroy(SArray* results) { static void indexInterResultsDestroy(SArray* results) {
@ -406,18 +406,18 @@ static int indexMergeFinalResults(SArray* interResults, EIndexOperatorType oType
return 0; return 0;
} }
static void indexMayMergeTempToFinalResult(SArray* result, TFileValue* tfv, SIdxTempResult* tr) { static void indexMayMergeTempToFinalResult(SArray* result, TFileValue* tfv, SIdxTRslt* tr) {
int32_t sz = taosArrayGetSize(result); int32_t sz = taosArrayGetSize(result);
if (sz > 0) { if (sz > 0) {
TFileValue* lv = taosArrayGetP(result, sz - 1); TFileValue* lv = taosArrayGetP(result, sz - 1);
if (tfv != NULL && strcmp(lv->colVal, tfv->colVal) != 0) { if (tfv != NULL && strcmp(lv->colVal, tfv->colVal) != 0) {
idxTempResultMergeTo(tr, lv->tableId); idxTRsltMergeTo(tr, lv->tableId);
idxTempResultClear(tr); idxTRsltClear(tr);
taosArrayPush(result, &tfv); taosArrayPush(result, &tfv);
} else if (tfv == NULL) { } else if (tfv == NULL) {
// handle last iterator // handle last iterator
idxTempResultMergeTo(tr, lv->tableId); idxTRsltMergeTo(tr, lv->tableId);
} else { } else {
// temp result saved in help // temp result saved in help
tfileValueDestroy(tfv); tfileValueDestroy(tfv);
@ -426,7 +426,7 @@ static void indexMayMergeTempToFinalResult(SArray* result, TFileValue* tfv, SIdx
taosArrayPush(result, &tfv); taosArrayPush(result, &tfv);
} }
} }
static void indexMergeCacheAndTFile(SArray* result, IterateValue* cv, IterateValue* tv, SIdxTempResult* tr) { static void indexMergeCacheAndTFile(SArray* result, IterateValue* cv, IterateValue* tv, SIdxTRslt* tr) {
char* colVal = (cv != NULL) ? cv->colVal : tv->colVal; char* colVal = (cv != NULL) ? cv->colVal : tv->colVal;
TFileValue* tfv = tfileValueCreate(colVal); TFileValue* tfv = tfileValueCreate(colVal);
@ -436,9 +436,9 @@ static void indexMergeCacheAndTFile(SArray* result, IterateValue* cv, IterateVal
uint64_t id = *(uint64_t*)taosArrayGet(cv->val, 0); uint64_t id = *(uint64_t*)taosArrayGet(cv->val, 0);
uint32_t ver = cv->ver; uint32_t ver = cv->ver;
if (cv->type == ADD_VALUE) { if (cv->type == ADD_VALUE) {
INDEX_MERGE_ADD_DEL(tr->deled, tr->added, id) INDEX_MERGE_ADD_DEL(tr->del, tr->add, id)
} else if (cv->type == DEL_VALUE) { } else if (cv->type == DEL_VALUE) {
INDEX_MERGE_ADD_DEL(tr->added, tr->deled, id) INDEX_MERGE_ADD_DEL(tr->add, tr->del, id)
} }
} }
if (tv != NULL) { if (tv != NULL) {
@ -491,7 +491,7 @@ int indexFlushCacheToTFile(SIndex* sIdx, void* cache) {
bool cn = cacheIter ? cacheIter->next(cacheIter) : false; bool cn = cacheIter ? cacheIter->next(cacheIter) : false;
bool tn = tfileIter ? tfileIter->next(tfileIter) : false; bool tn = tfileIter ? tfileIter->next(tfileIter) : false;
SIdxTempResult* tr = idxTempResultCreate(); SIdxTRslt* tr = idxTRsltCreate();
while (cn == true || tn == true) { while (cn == true || tn == true) {
IterateValue* cv = (cn == true) ? cacheIter->getValue(cacheIter) : NULL; IterateValue* cv = (cn == true) ? cacheIter->getValue(cacheIter) : NULL;
IterateValue* tv = (tn == true) ? tfileIter->getValue(tfileIter) : NULL; IterateValue* tv = (tn == true) ? tfileIter->getValue(tfileIter) : NULL;
@ -517,7 +517,7 @@ int indexFlushCacheToTFile(SIndex* sIdx, void* cache) {
} }
} }
indexMayMergeTempToFinalResult(result, NULL, tr); indexMayMergeTempToFinalResult(result, NULL, tr);
idxTempResultDestroy(tr); idxTRsltDestroy(tr);
int ret = indexGenTFile(sIdx, pCache, result); int ret = indexGenTFile(sIdx, pCache, result);
indexDestroyFinalResult(result); indexDestroyFinalResult(result);

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

@ -36,32 +36,31 @@ static char* indexCacheTermGet(const void* pData);
static MemTable* indexInternalCacheCreate(int8_t type); static MemTable* indexInternalCacheCreate(int8_t type);
static int32_t cacheSearchTerm(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchTerm(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchPrefix(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchPrefix(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchSuffix(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchSuffix(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchRegex(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchRegex(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchLessThan(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchLessThan(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchLessEqual(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchLessEqual(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchGreaterThan(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchGreaterThan(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchGreaterEqual(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchGreaterEqual(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchRange(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchRange(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
/*comm func of compare, used in (LE/LT/GE/GT compare)*/ /*comm func of compare, used in (LE/LT/GE/GT compare)*/
static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s, static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s, RangeType type);
RangeType type); static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchPrefix_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchPrefix_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchSuffix_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchSuffix_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchRegex_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchRegex_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchLessThan_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchLessThan_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchLessEqual_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchLessEqual_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchGreaterThan_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchGreaterThan_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchGreaterEqual_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchGreaterEqual_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s); static int32_t cacheSearchRange_JSON(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s);
static int32_t cacheSearchRange_JSON(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s, static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s,
RangeType type); RangeType type);
static int32_t (*cacheSearch[][QUERY_MAX])(void* cache, SIndexTerm* ct, SIdxTempResult* tr, STermValueType* s) = { static int32_t (*cacheSearch[][QUERY_MAX])(void* cache, SIndexTerm* ct, SIdxTRslt* tr, STermValueType* s) = {
{cacheSearchTerm, cacheSearchPrefix, cacheSearchSuffix, cacheSearchRegex, cacheSearchLessThan, cacheSearchLessEqual, {cacheSearchTerm, cacheSearchPrefix, cacheSearchSuffix, cacheSearchRegex, cacheSearchLessThan, cacheSearchLessEqual,
cacheSearchGreaterThan, cacheSearchGreaterEqual, cacheSearchRange}, cacheSearchGreaterThan, cacheSearchGreaterEqual, cacheSearchRange},
{cacheSearchTerm_JSON, cacheSearchPrefix_JSON, cacheSearchSuffix_JSON, cacheSearchRegex_JSON, {cacheSearchTerm_JSON, cacheSearchPrefix_JSON, cacheSearchSuffix_JSON, cacheSearchRegex_JSON,
@ -71,7 +70,7 @@ static int32_t (*cacheSearch[][QUERY_MAX])(void* cache, SIndexTerm* ct, SIdxTemp
static void doMergeWork(SSchedMsg* msg); static void doMergeWork(SSchedMsg* msg);
static bool indexCacheIteratorNext(Iterate* itera); static bool indexCacheIteratorNext(Iterate* itera);
static int32_t cacheSearchTerm(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchTerm(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
if (cache == NULL) { if (cache == NULL) {
return 0; return 0;
} }
@ -93,11 +92,11 @@ static int32_t cacheSearchTerm(void* cache, SIndexTerm* term, SIdxTempResult* tr
CacheTerm* c = (CacheTerm*)SL_GET_NODE_DATA(node); CacheTerm* c = (CacheTerm*)SL_GET_NODE_DATA(node);
if (0 == strcmp(c->colVal, pCt->colVal)) { if (0 == strcmp(c->colVal, pCt->colVal)) {
if (c->operaType == ADD_VALUE) { if (c->operaType == ADD_VALUE) {
INDEX_MERGE_ADD_DEL(tr->deled, tr->added, c->uid) INDEX_MERGE_ADD_DEL(tr->del, tr->add, c->uid)
// taosArrayPush(result, &c->uid); // taosArrayPush(result, &c->uid);
*s = kTypeValue; *s = kTypeValue;
} else if (c->operaType == DEL_VALUE) { } else if (c->operaType == DEL_VALUE) {
INDEX_MERGE_ADD_DEL(tr->added, tr->deled, c->uid) INDEX_MERGE_ADD_DEL(tr->add, tr->del, c->uid)
} }
} else { } else {
break; break;
@ -108,20 +107,19 @@ static int32_t cacheSearchTerm(void* cache, SIndexTerm* term, SIdxTempResult* tr
tSkipListDestroyIter(iter); tSkipListDestroyIter(iter);
return 0; return 0;
} }
static int32_t cacheSearchPrefix(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchPrefix(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
// impl later // impl later
return 0; return 0;
} }
static int32_t cacheSearchSuffix(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchSuffix(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
// impl later // impl later
return 0; return 0;
} }
static int32_t cacheSearchRegex(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchRegex(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
// impl later // impl later
return 0; return 0;
} }
static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s, static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s, RangeType type) {
RangeType type) {
if (cache == NULL) { if (cache == NULL) {
return 0; return 0;
} }
@ -148,11 +146,11 @@ static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* term, SIdxTempRes
TExeCond cond = cmpFn(c->colVal, pCt->colVal, pCt->colType); TExeCond cond = cmpFn(c->colVal, pCt->colVal, pCt->colType);
if (cond == MATCH) { if (cond == MATCH) {
if (c->operaType == ADD_VALUE) { if (c->operaType == ADD_VALUE) {
INDEX_MERGE_ADD_DEL(tr->deled, tr->added, c->uid) INDEX_MERGE_ADD_DEL(tr->del, tr->add, c->uid)
// taosArrayPush(result, &c->uid); // taosArrayPush(result, &c->uid);
*s = kTypeValue; *s = kTypeValue;
} else if (c->operaType == DEL_VALUE) { } else if (c->operaType == DEL_VALUE) {
INDEX_MERGE_ADD_DEL(tr->added, tr->deled, c->uid) INDEX_MERGE_ADD_DEL(tr->add, tr->del, c->uid)
} }
} else if (cond == CONTINUE) { } else if (cond == CONTINUE) {
continue; continue;
@ -164,20 +162,20 @@ static int32_t cacheSearchCompareFunc(void* cache, SIndexTerm* term, SIdxTempRes
tSkipListDestroyIter(iter); tSkipListDestroyIter(iter);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t cacheSearchLessThan(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchLessThan(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, term, tr, s, LT); return cacheSearchCompareFunc(cache, term, tr, s, LT);
} }
static int32_t cacheSearchLessEqual(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchLessEqual(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, term, tr, s, LE); return cacheSearchCompareFunc(cache, term, tr, s, LE);
} }
static int32_t cacheSearchGreaterThan(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchGreaterThan(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, term, tr, s, GT); return cacheSearchCompareFunc(cache, term, tr, s, GT);
} }
static int32_t cacheSearchGreaterEqual(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchGreaterEqual(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, term, tr, s, GE); return cacheSearchCompareFunc(cache, term, tr, s, GE);
} }
static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
if (cache == NULL) { if (cache == NULL) {
return 0; return 0;
} }
@ -205,11 +203,11 @@ static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* term, SIdxTempResul
if (0 == strcmp(c->colVal, pCt->colVal)) { if (0 == strcmp(c->colVal, pCt->colVal)) {
if (c->operaType == ADD_VALUE) { if (c->operaType == ADD_VALUE) {
INDEX_MERGE_ADD_DEL(tr->deled, tr->added, c->uid) INDEX_MERGE_ADD_DEL(tr->del, tr->add, c->uid)
// taosArrayPush(result, &c->uid); // taosArrayPush(result, &c->uid);
*s = kTypeValue; *s = kTypeValue;
} else if (c->operaType == DEL_VALUE) { } else if (c->operaType == DEL_VALUE) {
INDEX_MERGE_ADD_DEL(tr->added, tr->deled, c->uid) INDEX_MERGE_ADD_DEL(tr->add, tr->del, c->uid)
} }
} else { } else {
break; break;
@ -223,32 +221,32 @@ static int32_t cacheSearchTerm_JSON(void* cache, SIndexTerm* term, SIdxTempResul
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t cacheSearchPrefix_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchPrefix_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t cacheSearchSuffix_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchSuffix_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t cacheSearchRegex_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchRegex_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t cacheSearchLessThan_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchLessThan_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc_JSON(cache, term, tr, s, LT); return cacheSearchCompareFunc_JSON(cache, term, tr, s, LT);
} }
static int32_t cacheSearchLessEqual_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchLessEqual_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc_JSON(cache, term, tr, s, LE); return cacheSearchCompareFunc_JSON(cache, term, tr, s, LE);
} }
static int32_t cacheSearchGreaterThan_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchGreaterThan_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc_JSON(cache, term, tr, s, GT); return cacheSearchCompareFunc_JSON(cache, term, tr, s, GT);
} }
static int32_t cacheSearchGreaterEqual_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchGreaterEqual_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return cacheSearchCompareFunc_JSON(cache, term, tr, s, GE); return cacheSearchCompareFunc_JSON(cache, term, tr, s, GE);
} }
static int32_t cacheSearchRange_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchRange_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s, static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s,
RangeType type) { RangeType type) {
if (cache == NULL) { if (cache == NULL) {
return 0; return 0;
@ -290,11 +288,11 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTe
TExeCond cond = cmpFn(p + skip, term->colVal, dType); TExeCond cond = cmpFn(p + skip, term->colVal, dType);
if (cond == MATCH) { if (cond == MATCH) {
if (c->operaType == ADD_VALUE) { if (c->operaType == ADD_VALUE) {
INDEX_MERGE_ADD_DEL(tr->deled, tr->added, c->uid) INDEX_MERGE_ADD_DEL(tr->del, tr->add, c->uid)
// taosArrayPush(result, &c->uid); // taosArrayPush(result, &c->uid);
*s = kTypeValue; *s = kTypeValue;
} else if (c->operaType == DEL_VALUE) { } else if (c->operaType == DEL_VALUE) {
INDEX_MERGE_ADD_DEL(tr->added, tr->deled, c->uid) INDEX_MERGE_ADD_DEL(tr->add, tr->del, c->uid)
} }
} else if (cond == CONTINUE) { } else if (cond == CONTINUE) {
continue; continue;
@ -310,7 +308,7 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTe
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t cacheSearchRange(void* cache, SIndexTerm* term, SIdxTempResult* tr, STermValueType* s) { static int32_t cacheSearchRange(void* cache, SIndexTerm* term, SIdxTRslt* tr, STermValueType* s) {
// impl later // impl later
return 0; return 0;
} }
@ -569,7 +567,7 @@ int indexCacheDel(void* cache, const char* fieldValue, int32_t fvlen, uint64_t u
return 0; return 0;
} }
static int32_t indexQueryMem(MemTable* mem, SIndexTermQuery* query, SIdxTempResult* tr, STermValueType* s) { static int32_t indexQueryMem(MemTable* mem, SIndexTermQuery* query, SIdxTRslt* tr, STermValueType* s) {
if (mem == NULL) { if (mem == NULL) {
return 0; return 0;
} }
@ -583,7 +581,7 @@ static int32_t indexQueryMem(MemTable* mem, SIndexTermQuery* query, SIdxTempResu
return cacheSearch[0][qtype](mem, term, tr, s); return cacheSearch[0][qtype](mem, term, tr, s);
} }
} }
int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTempResult* result, STermValueType* s) { int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* result, STermValueType* s) {
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
if (cache == NULL) { if (cache == NULL) {
return 0; return 0;

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

@ -60,31 +60,31 @@ static void tfileGenFileFullName(char* fullname, const char* path, uint64_t s
/* /*
* search from tfile * search from tfile
*/ */
static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchLessThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchLessThan(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchLessEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchLessEqual(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchGreaterThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchGreaterThan(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchGreaterEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchGreaterEqual(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTempResult* tr, RangeType ctype); static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTRslt* tr, RangeType ctype);
static int32_t tfSearchTerm_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchTerm_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchPrefix_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchPrefix_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchSuffix_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchSuffix_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchRegex_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchRegex_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchLessThan_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchLessThan_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchLessEqual_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchLessEqual_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchGreaterThan_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchGreaterThan_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchGreaterEqual_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchGreaterEqual_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchRange_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr); static int32_t tfSearchRange_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr);
static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr, RangeType ctype); static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr, RangeType ctype);
static int32_t (*tfSearch[][QUERY_MAX])(void* reader, SIndexTerm* tem, SIdxTempResult* tr) = { static int32_t (*tfSearch[][QUERY_MAX])(void* reader, SIndexTerm* tem, SIdxTRslt* tr) = {
{tfSearchTerm, tfSearchPrefix, tfSearchSuffix, tfSearchRegex, tfSearchLessThan, tfSearchLessEqual, {tfSearchTerm, tfSearchPrefix, tfSearchSuffix, tfSearchRegex, tfSearchLessThan, tfSearchLessEqual,
tfSearchGreaterThan, tfSearchGreaterEqual, tfSearchRange}, tfSearchGreaterThan, tfSearchGreaterEqual, tfSearchRange},
{tfSearchTerm_JSON, tfSearchPrefix_JSON, tfSearchSuffix_JSON, tfSearchRegex_JSON, tfSearchLessThan_JSON, {tfSearchTerm_JSON, tfSearchPrefix_JSON, tfSearchSuffix_JSON, tfSearchRegex_JSON, tfSearchLessThan_JSON,
@ -220,7 +220,7 @@ void tfileReaderDestroy(TFileReader* reader) {
taosMemoryFree(reader); taosMemoryFree(reader);
} }
static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
int ret = 0; int ret = 0;
char* p = tem->colVal; char* p = tem->colVal;
uint64_t sz = tem->nColVal; uint64_t sz = tem->nColVal;
@ -243,7 +243,7 @@ static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
return 0; return 0;
} }
static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON); bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
char* p = tem->colVal; char* p = tem->colVal;
uint64_t sz = tem->nColVal; uint64_t sz = tem->nColVal;
@ -279,7 +279,7 @@ static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTempResult* tr)
} }
return 0; return 0;
} }
static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON); bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
int ret = 0; int ret = 0;
@ -298,7 +298,7 @@ static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTempResult* tr)
fstSliceDestroy(&key); fstSliceDestroy(&key);
return 0; return 0;
} }
static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { 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 = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
int ret = 0; int ret = 0;
@ -319,7 +319,7 @@ static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr)
return 0; return 0;
} }
static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTempResult* tr, RangeType type) { static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTRslt* tr, RangeType type) {
int ret = 0; int ret = 0;
char* p = tem->colVal; char* p = tem->colVal;
int skip = 0; int skip = 0;
@ -358,19 +358,19 @@ static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTempResult
fstStreamBuilderDestroy(sb); fstStreamBuilderDestroy(sb);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tfSearchLessThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchLessThan(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc(reader, tem, tr, LT); return tfSearchCompareFunc(reader, tem, tr, LT);
} }
static int32_t tfSearchLessEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchLessEqual(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc(reader, tem, tr, LE); return tfSearchCompareFunc(reader, tem, tr, LE);
} }
static int32_t tfSearchGreaterThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchGreaterThan(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc(reader, tem, tr, GT); return tfSearchCompareFunc(reader, tem, tr, GT);
} }
static int32_t tfSearchGreaterEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchGreaterEqual(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc(reader, tem, tr, GE); return tfSearchCompareFunc(reader, tem, tr, GE);
} }
static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON); bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
int ret = 0; int ret = 0;
char* p = tem->colVal; char* p = tem->colVal;
@ -399,7 +399,7 @@ static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTempResult* tr)
fstSliceDestroy(&key); fstSliceDestroy(&key);
return 0; return 0;
} }
static int32_t tfSearchTerm_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchTerm_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
int ret = 0; int ret = 0;
char* p = indexPackJsonData(tem); char* p = indexPackJsonData(tem);
int sz = strlen(p); int sz = strlen(p);
@ -424,36 +424,36 @@ static int32_t tfSearchTerm_JSON(void* reader, SIndexTerm* tem, SIdxTempResult*
// deprecate api // deprecate api
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tfSearchPrefix_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchPrefix_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
// impl later // impl later
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tfSearchSuffix_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchSuffix_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
// impl later // impl later
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tfSearchRegex_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchRegex_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
// impl later // impl later
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tfSearchLessThan_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchLessThan_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc_JSON(reader, tem, tr, LT); return tfSearchCompareFunc_JSON(reader, tem, tr, LT);
} }
static int32_t tfSearchLessEqual_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchLessEqual_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc_JSON(reader, tem, tr, LE); return tfSearchCompareFunc_JSON(reader, tem, tr, LE);
} }
static int32_t tfSearchGreaterThan_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchGreaterThan_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc_JSON(reader, tem, tr, GT); return tfSearchCompareFunc_JSON(reader, tem, tr, GT);
} }
static int32_t tfSearchGreaterEqual_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchGreaterEqual_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
return tfSearchCompareFunc_JSON(reader, tem, tr, GE); return tfSearchCompareFunc_JSON(reader, tem, tr, GE);
} }
static int32_t tfSearchRange_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr) { static int32_t tfSearchRange_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
// impl later // impl later
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTempResult* tr, RangeType ctype) { static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTRslt* tr, RangeType ctype) {
int ret = 0; int ret = 0;
int skip = 0; int skip = 0;
@ -501,7 +501,7 @@ static int32_t tfSearchCompareFunc_JSON(void* reader, SIndexTerm* tem, SIdxTempR
fstStreamBuilderDestroy(sb); fstStreamBuilderDestroy(sb);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTempResult* tr) { int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTRslt* tr) {
SIndexTerm* term = query->term; SIndexTerm* term = query->term;
EIndexQueryType qtype = query->qType; EIndexQueryType qtype = query->qType;
int ret = 0; int ret = 0;
@ -673,7 +673,7 @@ void indexTFileDestroy(IndexTFile* tfile) {
taosMemoryFree(tfile); taosMemoryFree(tfile);
} }
int indexTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTempResult* result) { int indexTFileSearch(void* tfile, SIndexTermQuery* query, SIdxTRslt* result) {
int ret = -1; int ret = -1;
if (tfile == NULL) { if (tfile == NULL) {
return ret; return ret;

78
source/libs/index/src/indexUtil.c
View File

@ -36,24 +36,24 @@ static int iBinarySearch(SArray *arr, int s, int e, uint64_t k) {
return s; return s;
} }
void iIntersection(SArray *inters, SArray *final) { void iIntersection(SArray *in, SArray *out) {
int32_t sz = (int32_t)taosArrayGetSize(inters); int32_t sz = (int32_t)taosArrayGetSize(in);
if (sz <= 0) { if (sz <= 0) {
return; return;
} }
MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex)); MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex));
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
SArray *t = taosArrayGetP(inters, i); SArray *t = taosArrayGetP(in, i);
mi[i].len = (int32_t)taosArrayGetSize(t); mi[i].len = (int32_t)taosArrayGetSize(t);
mi[i].idx = 0; mi[i].idx = 0;
} }
SArray *base = taosArrayGetP(inters, 0); SArray *base = taosArrayGetP(in, 0);
for (int i = 0; i < taosArrayGetSize(base); i++) { for (int i = 0; i < taosArrayGetSize(base); i++) {
uint64_t tgt = *(uint64_t *)taosArrayGet(base, i); uint64_t tgt = *(uint64_t *)taosArrayGet(base, i);
bool has = true; bool has = true;
for (int j = 1; j < taosArrayGetSize(inters); j++) { for (int j = 1; j < taosArrayGetSize(in); j++) {
SArray *oth = taosArrayGetP(inters, j); SArray *oth = taosArrayGetP(in, j);
int mid = iBinarySearch(oth, mi[j].idx, mi[j].len - 1, tgt); int mid = iBinarySearch(oth, mi[j].idx, mi[j].len - 1, tgt);
if (mid >= 0 && mid < mi[j].len) { if (mid >= 0 && mid < mi[j].len) {
uint64_t val = *(uint64_t *)taosArrayGet(oth, mid); uint64_t val = *(uint64_t *)taosArrayGet(oth, mid);
@ -64,24 +64,24 @@ void iIntersection(SArray *inters, SArray *final) {
} }
} }
if (has == true) { if (has == true) {
taosArrayPush(final, &tgt); taosArrayPush(out, &tgt);
} }
} }
taosMemoryFreeClear(mi); taosMemoryFreeClear(mi);
} }
void iUnion(SArray *inters, SArray *final) { void iUnion(SArray *in, SArray *out) {
int32_t sz = (int32_t)taosArrayGetSize(inters); int32_t sz = (int32_t)taosArrayGetSize(in);
if (sz <= 0) { if (sz <= 0) {
return; return;
} }
if (sz == 1) { if (sz == 1) {
taosArrayAddAll(final, taosArrayGetP(inters, 0)); taosArrayAddAll(out, taosArrayGetP(in, 0));
return; return;
} }
MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex)); MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex));
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
SArray *t = taosArrayGetP(inters, i); SArray *t = taosArrayGetP(in, i);
mi[i].len = (int32_t)taosArrayGetSize(t); mi[i].len = (int32_t)taosArrayGetSize(t);
mi[i].idx = 0; mi[i].idx = 0;
} }
@ -90,7 +90,7 @@ void iUnion(SArray *inters, SArray *final) {
int mIdx = -1; int mIdx = -1;
for (int j = 0; j < sz; j++) { for (int j = 0; j < sz; j++) {
SArray *t = taosArrayGetP(inters, j); SArray *t = taosArrayGetP(in, j);
if (mi[j].idx >= mi[j].len) { if (mi[j].idx >= mi[j].len) {
continue; continue;
} }
@ -102,13 +102,13 @@ void iUnion(SArray *inters, SArray *final) {
} }
if (mIdx != -1) { if (mIdx != -1) {
mi[mIdx].idx++; mi[mIdx].idx++;
if (taosArrayGetSize(final) > 0) { if (taosArrayGetSize(out) > 0) {
uint64_t lVal = *(uint64_t *)taosArrayGetLast(final); uint64_t lVal = *(uint64_t *)taosArrayGetLast(out);
if (lVal == mVal) { if (lVal == mVal) {
continue; continue;
} }
} }
taosArrayPush(final, &mVal); taosArrayPush(out, &mVal);
} else { } else {
break; break;
} }
@ -158,46 +158,44 @@ int verdataCompare(const void *a, const void *b) {
return cmp; return cmp;
} }
SIdxTempResult *idxTempResultCreate() { SIdxTRslt *idxTRsltCreate() {
SIdxTempResult *tr = taosMemoryCalloc(1, sizeof(SIdxTempResult)); SIdxTRslt *tr = taosMemoryCalloc(1, sizeof(SIdxTRslt));
tr->total = taosArrayInit(4, sizeof(uint64_t)); tr->total = taosArrayInit(4, sizeof(uint64_t));
tr->added = taosArrayInit(4, sizeof(uint64_t)); tr->add = taosArrayInit(4, sizeof(uint64_t));
tr->deled = taosArrayInit(4, sizeof(uint64_t)); tr->del = taosArrayInit(4, sizeof(uint64_t));
return tr; return tr;
} }
void idxTempResultClear(SIdxTempResult *tr) { void idxTRsltClear(SIdxTRslt *tr) {
if (tr == NULL) { if (tr == NULL) {
return; return;
} }
taosArrayClear(tr->total); taosArrayClear(tr->total);
taosArrayClear(tr->added); taosArrayClear(tr->add);
taosArrayClear(tr->deled); taosArrayClear(tr->del);
} }
void idxTempResultDestroy(SIdxTempResult *tr) { void idxTRsltDestroy(SIdxTRslt *tr) {
if (tr == NULL) { if (tr == NULL) {
return; return;
} }
taosArrayDestroy(tr->total); taosArrayDestroy(tr->total);
taosArrayDestroy(tr->added); taosArrayDestroy(tr->add);
taosArrayDestroy(tr->deled); taosArrayDestroy(tr->del);
} }
void idxTempResultMergeTo(SIdxTempResult *tr, SArray *result) { void idxTRsltMergeTo(SIdxTRslt *tr, SArray *result) {
taosArraySort(tr->total, uidCompare); taosArraySort(tr->total, uidCompare);
taosArraySort(tr->added, uidCompare); taosArraySort(tr->add, uidCompare);
taosArraySort(tr->deled, uidCompare); taosArraySort(tr->del, uidCompare);
SArray *arrs = taosArrayInit(2, sizeof(void *)); if (taosArrayGetSize(tr->total) == 0 || taosArrayGetSize(tr->add) == 0) {
taosArrayPush(arrs, &tr->total); SArray *t = taosArrayGetSize(tr->total) == 0 ? tr->add : tr->total;
taosArrayPush(arrs, &tr->added); taosArrayAddAll(result, t);
} else {
iUnion(arrs, result); SArray *arrs = taosArrayInit(2, sizeof(void *));
taosArrayDestroy(arrs); taosArrayPush(arrs, &tr->total);
taosArrayPush(arrs, &tr->add);
indexError("tmp result: total: %d, added: %d, del: %d", (int)taosArrayGetSize(tr->total), iUnion(arrs, result);
(int)taosArrayGetSize(tr->added), (int)taosArrayGetSize(tr->deled)); taosArrayDestroy(arrs);
if (taosArrayGetSize(tr->added) != 0 && taosArrayGetSize(result) == 0) {
indexError("except result: %d", (int)(taosArrayGetSize(result)));
} }
iExcept(result, tr->deled); iExcept(result, tr->del);
} }

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

@ -411,12 +411,12 @@ class TFileObj {
// //
// //
} }
SIdxTempResult* tr = idxTempResultCreate(); SIdxTRslt* tr = idxTRsltCreate();
int ret = tfileReaderSearch(reader_, query, tr); int ret = tfileReaderSearch(reader_, query, tr);
idxTempResultMergeTo(tr, result); idxTRsltMergeTo(tr, result);
idxTempResultDestroy(tr); idxTRsltDestroy(tr);
return ret; return ret;
} }
~TFileObj() { ~TFileObj() {
@ -531,11 +531,11 @@ class CacheObj {
indexCacheDebug(cache); indexCacheDebug(cache);
} }
int Get(SIndexTermQuery* query, int16_t colId, int32_t version, SArray* result, STermValueType* s) { int Get(SIndexTermQuery* query, int16_t colId, int32_t version, SArray* result, STermValueType* s) {
SIdxTempResult* tr = idxTempResultCreate(); SIdxTRslt* tr = idxTRsltCreate();
int ret = indexCacheSearch(cache, query, tr, s); int ret = indexCacheSearch(cache, query, tr, s);
idxTempResultMergeTo(tr, result); idxTRsltMergeTo(tr, result);
idxTempResultDestroy(tr); idxTRsltDestroy(tr);
if (ret != 0) { if (ret != 0) {
std::cout << "failed to get from cache:" << ret << std::endl; std::cout << "failed to get from cache:" << ret << std::endl;

16
source/libs/index/test/utilUT.cc
View File

@ -338,12 +338,22 @@ TEST_F(UtilEnv, testFill) {
} }
} }
TEST_F(UtilEnv, TempResult) { TEST_F(UtilEnv, TempResult) {
SIdxTempResult *relt = idxTempResultCreate(); SIdxTRslt *relt = idxTRsltCreate();
SArray *f = taosArrayInit(0, sizeof(uint64_t)); SArray *f = taosArrayInit(0, sizeof(uint64_t));
uint64_t val = UINT64_MAX - 1; uint64_t val = UINT64_MAX - 1;
taosArrayPush(relt->added, &val); taosArrayPush(relt->add, &val);
idxTempResultMergeTo(relt, f); idxTRsltMergeTo(relt, f);
EXPECT_EQ(taosArrayGetSize(f), 1);
}
TEST_F(UtilEnv, TempResultExcept) {
SIdxTRslt *relt = idxTRsltCreate();
SArray *f = taosArrayInit(0, sizeof(uint64_t));
uint64_t val = UINT64_MAX;
taosArrayPush(relt->add, &val);
idxTRsltMergeTo(relt, f);
EXPECT_EQ(taosArrayGetSize(f), 1); EXPECT_EQ(taosArrayGetSize(f), 1);
} }