enh(index): add prefix query

include/libs/index/index.h

@@ -17,6 +17,7 @@
 #define _TD_INDEX_H_
 
 #include "os.h"
+#include "taoserror.h"
 #include "tarray.h"
 
 #ifdef __cplusplus
@@ -41,7 +42,8 @@ typedef enum {
   UPDATE_VALUE,  // update index column value
   ADD_INDEX,     // add index on specify column
   DROP_INDEX,    // drop existed index
-  DROP_SATBLE    // drop stable
+  DROP_SATBLE,   // drop stable
+  DEFAULT        // query
 } SIndexOperOnColumn;
 
 typedef enum { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType;

source/libs/executor/src/indexoperator.c

@@ -55,11 +55,11 @@ typedef struct SIFParam {
   SArray *result;
   char *  condValue;
 
+  uint8_t  colValType;
   col_id_t colId;
-
-  int64_t suid;  // add later
-  char    dbName[TSDB_DB_NAME_LEN];
-  char    colName[TSDB_COL_NAME_LEN];
+  int64_t  suid;  // add later
+  char     dbName[TSDB_DB_NAME_LEN];
+  char     colName[TSDB_COL_NAME_LEN];
 } SIFParam;
 
 typedef int32_t (*sif_func_t)(SIFParam *left, SIFParam *rigth, SIFParam *output);
@@ -145,10 +145,11 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
       SColumnNode *cn = (SColumnNode *)node;
       /*only support tag column*/
       SIF_ERR_RET(sifValidateColumn(cn));
+
       param->colId = cn->colId;
+      param->colValType = cn->node.resType.type;
       memcpy(param->dbName, cn->dbName, sizeof(cn->dbName));
       memcpy(param->colName, cn->colName, sizeof(cn->colName));
-
       break;
     }
     case QUERY_NODE_NODE_LIST: {
@@ -231,61 +232,68 @@ static int32_t sifExecFunction(SFunctionNode *node, SIFCtx *ctx, SIFParam *outpu
   qError("index-filter not support buildin function");
   return TSDB_CODE_QRY_INVALID_INPUT;
 }
-static int32_t sifIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFParam *output) {
-  SIndexMultiTermQuery *mq = indexMultiTermQueryCreate(MUST);
-  return TSDB_CODE_SUCCESS;
+static int32_t sifDoIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFParam *output) {
+  SIndexTerm *tm = indexTermCreate(left->suid, DEFAULT, left->colValType, left->colName, strlen(left->colName),
+                                   right->condValue, strlen(right->condValue));
+  if (tm == NULL) {
+    return TSDB_CODE_QRY_OUT_OF_MEMORY;
+  }
+
+  SIndexMultiTermQuery *mtm = indexMultiTermQueryCreate(MUST);
+  indexMultiTermQueryAdd(mtm, tm, QUERY_TERM);
+  return indexSearch(NULL, mtm, output->result);
 }
 
 static int32_t sifLessThanFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_LOWER_THAN;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifLessEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_LOWER_EQUAL;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 
 static int32_t sifGreaterThanFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_GREATER_THAN;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifGreaterEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_GREATER_EQUAL;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 
 static int32_t sifEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_EQUAL;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifNotEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_NOT_EQUAL;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifInFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_IN;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifNotInFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_NOT_IN;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifLikeFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_LIKE;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifNotLikeFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_NOT_LIKE;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 
 static int32_t sifMatchFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_MATCH;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifNotMatchFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   int id = OP_TYPE_NMATCH;
-  return sifIndex(left, right, id, output);
+  return sifDoIndex(left, right, id, output);
 }
 static int32_t sifDefaultFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   // add more except
@@ -460,6 +468,7 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) {
     qError("index-filter failed to taosHashInit");
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
+
   nodesWalkExprPostOrder(pNode, sifCalcWalker, &ctx);
   SIF_ERR_RET(ctx.code);
 

source/libs/index/src/index.c

@@ -175,55 +175,19 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
   return 0;
 }
 int indexSearch(SIndex* index, SIndexMultiTermQuery* multiQuerys, SArray* result) {
-#ifdef USE_LUCENE
-  EIndexOperatorType opera = multiQuerys->opera;
-
-  int    nQuery = taosArrayGetSize(multiQuerys->query);
-  char** fields = taosMemoryMalloc(sizeof(char*) * nQuery);
-  char** keys = taosMemoryMalloc(sizeof(char*) * nQuery);
-  int*   types = taosMemoryMalloc(sizeof(int) * nQuery);
-
-  for (int i = 0; i < nQuery; i++) {
-    SIndexTermQuery* p = taosArrayGet(multiQuerys->query, i);
-    SIndexTerm*      term = p->field_value;
-
-    fields[i] = taosMemoryCalloc(1, term->nKey + 1);
-    keys[i] = taosMemoryCalloc(1, term->nVal + 1);
-
-    memcpy(fields[i], term->key, term->nKey);
-    memcpy(keys[i], term->val, term->nVal);
-    types[i] = (int)(p->type);
-  }
-  int* tResult = NULL;
-  int  tsz = 0;
-  index_multi_search(index->index, (const char**)fields, (const char**)keys, types, nQuery, opera, &tResult, &tsz);
-
-  for (int i = 0; i < tsz; i++) {
-    taosArrayPush(result, &tResult[i]);
-  }
-
-  for (int i = 0; i < nQuery; i++) {
-    taosMemoryFree(fields[i]);
-    taosMemoryFree(keys[i]);
-  }
-  taosMemoryFree(fields);
-  taosMemoryFree(keys);
-  taosMemoryFree(types);
-#endif
-
 #ifdef USE_INVERTED_INDEX
   EIndexOperatorType opera = multiQuerys->opera;  // relation of querys
 
-  SArray* interResults = taosArrayInit(4, POINTER_BYTES);
+  SArray* iRslts = taosArrayInit(4, POINTER_BYTES);
   int     nQuery = taosArrayGetSize(multiQuerys->query);
   for (size_t i = 0; i < nQuery; i++) {
-    SIndexTermQuery* qTerm = taosArrayGet(multiQuerys->query, i);
-    SArray*          tResult = NULL;
-    indexTermSearch(index, qTerm, &tResult);
-    taosArrayPush(interResults, (void*)&tResult);
+    SIndexTermQuery* qterm = taosArrayGet(multiQuerys->query, i);
+    SArray*          trslt = NULL;
+    indexTermSearch(index, qterm, &trslt);
+    taosArrayPush(iRslts, (void*)&trslt);
   }
-  indexMergeFinalResults(interResults, opera, result);
-  indexInterResultsDestroy(interResults);
+  indexMergeFinalResults(iRslts, opera, result);
+  indexInterResultsDestroy(iRslts);
 
 #endif
   return 0;

source/libs/index/src/indexCache.c

@@ -34,9 +34,38 @@ static char*   indexCacheTermGet(const void* pData);
 
 static MemTable* indexInternalCacheCreate(int8_t type);
 
+static int32_t cacheSearchTerm(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
+static int32_t cacheSearchPrefix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
+static int32_t cacheSearchSuffix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
+static int32_t cacheSearchRegex(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
+static int32_t cacheSearchRange(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
+
+static int32_t (*cacheSearch[])(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) = {
+    cacheSearchTerm, cacheSearchPrefix, cacheSearchSuffix, cacheSearchRegex, cacheSearchRange};
+
 static void doMergeWork(SSchedMsg* msg);
 static bool indexCacheIteratorNext(Iterate* itera);
 
+static int32_t cacheSearchTerm(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
+  // impl later
+  return 0;
+}
+static int32_t cacheSearchPrefix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
+  // impl later
+  return 0;
+}
+static int32_t cacheSearchSuffix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
+  // impl later
+  return 0;
+}
+static int32_t cacheSearchRegex(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
+  // impl later
+  return 0;
+}
+static int32_t cacheSearchRange(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
+  // impl later
+  return 0;
+}
 static IterateValue* indexCacheIteratorGetValue(Iterate* iter);
 
 IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type) {

source/libs/index/src/indexTfile.c

@@ -57,6 +57,17 @@ static int     tfileCompare(const void* a, const void* b);
 static int     tfileParseFileName(const char* filename, uint64_t* suid, char* col, int* version);
 static void    tfileGenFileName(char* filename, uint64_t suid, const char* col, int version);
 static void    tfileGenFileFullName(char* fullname, const char* path, uint64_t suid, const char* col, int32_t version);
+/*
+ * search from  tfile
+ */
+static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
+static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
+static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
+static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
+static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
+
+static int32_t (*tfSearch[])(void* reader, SIndexTerm* tem, SIdxTempResult* tr) = {
+    tfSearchTerm, tfSearchPrefix, tfSearchSuffix, tfSearchRegex, tfSearchRange};
 
 TFileCache* tfileCacheCreate(const char* path) {
   TFileCache* tcache = taosMemoryCalloc(1, sizeof(TFileCache));
@@ -183,59 +194,153 @@ void tfileReaderDestroy(TFileReader* reader) {
   writerCtxDestroy(reader->ctx, reader->remove);
   taosMemoryFree(reader);
 }
+static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
+  bool     hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
+  int      ret = 0;
+  char*    p = tem->colVal;
+  uint64_t sz = tem->nColVal;
+  if (hasJson) {
+    p = indexPackJsonData(tem);
+    sz = strlen(p);
+  }
+  int64_t  st = taosGetTimestampUs();
+  FstSlice key = fstSliceCreate(p, sz);
+  uint64_t offset;
+  if (fstGet(((TFileReader*)reader)->fst, &key, &offset)) {
+    int64_t et = taosGetTimestampUs();
+    int64_t cost = et - st;
+    indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
+              tem->suid, tem->colName, tem->colVal, cost);
+
+    ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total);
+    cost = taosGetTimestampUs() - et;
+    indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", tem->suid,
+              tem->colName, tem->colVal, cost);
+  }
+  if (hasJson) {
+    taosMemoryFree(p);
+  }
+  fstSliceDestroy(&key);
+  return 0;
+}
 
-int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTempResult* tr) {
-  SIndexTerm*     term = query->term;
-  bool            hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON);
-  EIndexQueryType qtype = query->qType;
+static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
+  bool     hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
+  char*    p = tem->colVal;
+  uint64_t sz = tem->nColVal;
+  if (hasJson) {
+    p = indexPackJsonData(tem);
+    sz = strlen(p);
+  }
 
-  // SArray* result = taosArrayInit(16, sizeof(uint64_t));
-  int ret = -1;
-  // refactor to callback later
-  if (qtype == QUERY_TERM) {
-    uint64_t offset;
-    char*    p = term->colVal;
-    uint64_t sz = term->nColVal;
-    if (hasJson) {
-      p = indexPackJsonData(term);
-      sz = strlen(p);
-    }
-    int64_t  st = taosGetTimestampUs();
-    FstSlice key = fstSliceCreate(p, sz);
-    if (fstGet(reader->fst, &key, &offset)) {
-      int64_t et = taosGetTimestampUs();
-      int64_t cost = et - st;
-      indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
-                term->suid, term->colName, term->colVal, cost);
-
-      ret = tfileReaderLoadTableIds(reader, offset, tr->total);
-      cost = taosGetTimestampUs() - et;
-      indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", term->suid,
-                term->colName, term->colVal, cost);
-    } else {
-      indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found table info in tindex", term->suid, term->colName,
-                term->colVal);
-    }
-    fstSliceDestroy(&key);
-    if (hasJson) {
-      taosMemoryFree(p);
+  SArray* offsets = taosArrayInit(16, sizeof(uint64_t));
+
+  AutomationCtx*         ctx = automCtxCreate((void*)p, AUTOMATION_PREFIX);
+  FstStreamBuilder*      sb = fstSearch(((TFileReader*)reader)->fst, ctx);
+  StreamWithState*       st = streamBuilderIntoStream(sb);
+  StreamWithStateResult* rt = NULL;
+  while ((rt = streamWithStateNextWith(st, NULL)) != NULL) {
+    taosArrayPush(offsets, &(rt->out.out));
+    swsResultDestroy(rt);
+  }
+  streamWithStateDestroy(st);
+  fstStreamBuilderDestroy(sb);
+
+  int32_t ret = 0;
+  for (int i = 0; i < taosArrayGetSize(offsets); i++) {
+    uint64_t offset = *(uint64_t*)taosArrayGet(offsets, i);
+    ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total);
+    if (ret != 0) {
+      indexError("failed to find target tablelist");
+      return TSDB_CODE_TDB_FILE_CORRUPTED;
     }
-  } else if (qtype == QUERY_PREFIX) {
-    // handle later
-    //
-  } else if (qtype == QUERY_SUFFIX) {
-    // handle later
-  } else if (qtype == QUERY_REGEX) {
-    // handle later
-  } else if (qtype == QUERY_RANGE) {
-    // handle later
   }
-  tfileReaderUnRef(reader);
+  if (hasJson) {
+    taosMemoryFree(p);
+  }
+  return 0;
+}
+static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
+  bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
 
-  // taosArrayAddAll(tr->total, result);
-  // taosArrayDestroy(result);
+  int      ret = 0;
+  char*    p = tem->colVal;
+  uint64_t sz = tem->nColVal;
+  if (hasJson) {
+    p = indexPackJsonData(tem);
+    sz = strlen(p);
+  }
+  int64_t  st = taosGetTimestampUs();
+  FstSlice key = fstSliceCreate(p, sz);
+  /*impl later*/
+  if (hasJson) {
+    taosMemoryFree(p);
+  }
+  fstSliceDestroy(&key);
+  return 0;
+}
+static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
+  bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
+
+  int      ret = 0;
+  char*    p = tem->colVal;
+  uint64_t sz = tem->nColVal;
+  if (hasJson) {
+    p = indexPackJsonData(tem);
+    sz = strlen(p);
+  }
+  int64_t  st = taosGetTimestampUs();
+  FstSlice key = fstSliceCreate(p, sz);
+  /*impl later*/
+
+  if (hasJson) {
+    taosMemoryFree(p);
+  }
+  fstSliceDestroy(&key);
+  return 0;
+}
+static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
+  bool     hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
+  int      ret = 0;
+  char*    p = tem->colVal;
+  uint64_t sz = tem->nColVal;
+  if (hasJson) {
+    p = indexPackJsonData(tem);
+    sz = strlen(p);
+  }
+  int64_t  st = taosGetTimestampUs();
+  FstSlice key = fstSliceCreate(p, sz);
+  // uint64_t offset;
+  // if (fstGet(((TFileReader*)reader)->fst, &key, &offset)) {
+  //  int64_t et = taosGetTimestampUs();
+  //  int64_t cost = et - st;
+  //  indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
+  //            tem->suid, tem->colName, tem->colVal, cost);
+
+  //  ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total);
+  //  cost = taosGetTimestampUs() - et;
+  //  indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", tem->suid,
+  //            tem->colName, tem->colVal, cost);
+  //}
+  if (hasJson) {
+    taosMemoryFree(p);
+  }
+  fstSliceDestroy(&key);
+  return 0;
+}
 
-  return ret;
+int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTempResult* tr) {
+  SIndexTerm*     term = query->term;
+  EIndexQueryType qtype = query->qType;
+  if (qtype >= sizeof(tfSearch) / sizeof(tfSearch[0])) {
+    indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found table info in tindex", term->suid, term->colName,
+              term->colVal);
+    return -1;
+  } else {
+    return tfSearch[qtype](reader, term, tr);
+  }
+  tfileReaderUnRef(reader);
+  return 0;
 }
 
 TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int32_t version, const char* colName, uint8_t colType) {