Merge pull request #9187 from taosdata/feature/index_cache

adjust index interface

cmake/cmake.options

@@ -37,6 +37,7 @@ option(
     off 
 )
 
+
 option(
     BUILD_WITH_NURAFT
     "If build with NuRaft" 
@@ -54,3 +55,9 @@ option(
     "If use doxygen build documents" 
     OFF
 )
+
+option(
+   BUILD_WITH_INVERTEDINDEX
+   "If use invertedIndex" 
+   ON 
+)

include/libs/index/index.h

@@ -24,6 +24,7 @@ extern "C" {
 #endif
 
 typedef struct SIndex SIndex;
+typedef struct SIndexTerm SIndexTerm;
 typedef struct SIndexOpts SIndexOpts;
 typedef struct SIndexMultiTermQuery SIndexMultiTermQuery;
 typedef struct SArray               SIndexMultiTerm;
@@ -35,7 +36,7 @@ typedef enum  {
    ADD_INDEX,    // add index on specify column
    DROP_INDEX,   // drop existed index 
    DROP_SATBLE   // drop stable 
-} SIndexColumnType;
+} SIndexOperOnColumn;
 
 typedef enum  { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType;
 typedef enum  { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2,QUERY_REGEX = 3} EIndexQueryType;
@@ -45,12 +46,12 @@ typedef enum  { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2,QUERY_REGEX =
 */
 SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType oper);
 void            indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery);
-int             indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, const char *field, int32_t nFields, const char *value, int32_t nValue, EIndexQueryType type);
+int             indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EIndexQueryType type);
 /* 
  * @param:    
  * @param:
  */
-SIndex* indexOpen(SIndexOpts *opt, const char *path);
+int   indexOpen(SIndexOpts *opt, const char *path, SIndex **index);
 void  indexClose(SIndex *index);
 int   indexPut(SIndex *index,    SIndexMultiTerm *terms, int uid);
 int   indexDelete(SIndex *index, SIndexMultiTermQuery *query); 
@@ -61,8 +62,8 @@ int   indexRebuild(SIndex *index, SIndexOpts *opt);
  * @param
  */
 SIndexMultiTerm *indexMultiTermCreate(); 
-int     indexMultiTermAdd(SIndexMultiTerm *terms, const char *field, int32_t nFields, const char *value, int32_t nValue);
-void    indexMultiTermDestroy(SIndexMultiTerm *terms);
+int indexMultiTermAdd(SIndexMultiTerm  *terms, SIndexTerm *term);
+void indexMultiTermDestroy(SIndexMultiTerm *terms);
 /*
  * @param: 
  * @param:
@@ -70,6 +71,17 @@ void    indexMultiTermDestroy(SIndexMultiTerm *terms);
 SIndexOpts *indexOptsCreate();
 void       indexOptsDestroy(SIndexOpts *opts);
 
+
+/*
+ * @param:
+ * @param:
+ */
+
+SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn operType, uint8_t colType, 
+                            const char *colName, int32_t nColName, const char *colVal, int32_t nColVal);
+void        indexTermDestroy(SIndexTerm *p);
+
+
 #ifdef __cplusplus
 }
 #endif

source/libs/index/CMakeLists.txt

@@ -22,9 +22,13 @@ if (${BUILD_WITH_LUCENE})
     index
     PUBLIC lucene++
   )
-
 endif(${BUILD_WITH_LUCENE})
 
+if (${BUILD_WITH_INVERTEDINDEX})
+  add_definitions(-DUSE_INVERTED_INDEX) 
+endif(${BUILD_WITH_INVERTEDINDEX})
+
+
 if (${BUILD_TEST})
   add_subdirectory(test)
 endif(${BUILD_TEST})

source/libs/index/inc/indexInt.h

@@ -37,10 +37,10 @@ struct SIndex {
 #endif  
  void     *cache;
  void     *tindex; 
- SHashObj *fieldObj;// < field name, field id> 
+ SHashObj *colObj;// < field name, field id> 
  
  int64_t  suid;      // current super table id, -1 is normal table    
- int      fieldId;  // field id allocated to cache
+ int      colId;  // field id allocated to cache
  int32_t  cVersion; // current version allocated to cache 
  pthread_mutex_t mtx;
 };   
@@ -60,22 +60,21 @@ struct SIndexMultiTermQuery {
 
 // field and key;
 typedef struct SIndexTerm {
-  uint8_t type;  // term data type, str/interger/json
-  char    *key;
-  int32_t nKey;
-  char    *val;
-  int32_t nVal;
+  int64_t           suid; 
+  SIndexOperOnColumn operType; // oper type, add/del/update
+  uint8_t            colType;  // term data type, str/interger/json
+  char    *colName;
+  int32_t nColName;
+  char    *colVal;
+  int32_t nColVal;
 } SIndexTerm;
 
 typedef struct SIndexTermQuery {
-  SIndexTerm*     field_value;
-  EIndexQueryType type;
+  SIndexTerm*     term;
+  EIndexQueryType qType;
 } SIndexTermQuery;
 
 
-SIndexTerm *indexTermCreate(const char *key, int32_t nKey, const char *val, int32_t nVal);
-void        indexTermDestroy(SIndexTerm *p);
-
 
 #define indexFatal(...) do { if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("index FATAL ", 255, __VA_ARGS__); }}     while(0)
 #define indexError(...) do { if (sDebugFlag & DEBUG_ERROR) { taosPrintLog("index ERROR ", 255, __VA_ARGS__); }}     while(0)

source/libs/index/inc/index_cache.h

@@ -38,13 +38,13 @@ typedef struct IndexCache {
 // 
 IndexCache *indexCacheCreate();
 
-void indexCacheDestroy(IndexCache *cache);
+void indexCacheDestroy(void *cache);
 
-int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue,  int32_t fvLen, 
+int indexCachePut(void *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue,  int32_t fvLen, 
               uint32_t version, uint64_t uid, int8_t operType);
 
-int indexCacheGet(IndexCache *cache, uint64_t *rst);
-int indexCacheSearch(IndexCache *cache, SIndexMultiTermQuery *query, SArray *result);
+int indexCacheGet(void *cache, uint64_t *rst);
+int indexCacheSearch(void *cache, SIndexMultiTermQuery *query, SArray *result);
 
 #ifdef __cplusplus
 }

source/libs/index/src/index.c

@@ -22,11 +22,10 @@
 #endif
 
 
-typedef struct SIdxFieldInfo {
-  int fieldId;     // generated by index internal  
+typedef struct SIdxColInfo {
+  int colId;     // generated by index internal  
   int cVersion;
-  int type; // field type 
-} SIdxFieldInfo; 
+} SIdxColInfo; 
 
 static pthread_once_t isInit = PTHREAD_ONCE_INIT;
 static void indexInit();
@@ -38,22 +37,25 @@ static int indexMergeCacheIntoTindex(struct SIndex *sIdx) {
   indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid); 
   return 0;
 }
-SIndex *indexOpen(SIndexOpts *opts, const char *path) {
+int indexOpen(SIndexOpts *opts, const char *path, SIndex **index) {
   pthread_once(&isInit, indexInit);
   SIndex *sIdx = calloc(1, sizeof(SIndex));
+  if (sIdx == NULL) { return -1; }
 
 #ifdef USE_LUCENE  
   index_t *index = index_open(path);      
   sIdx->index = index;
 #endif
-  
+
   sIdx->cache    = (void*)indexCacheCreate();     
   sIdx->tindex   = NULL;
-  sIdx->fieldObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
-  sIdx->fieldId  = 1; 
+  sIdx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
+  sIdx->colId  = 1; 
   sIdx->cVersion = 1; 
   pthread_mutex_init(&sIdx->mtx, NULL);
-  return sIdx;  
+
+  *index = sIdx; 
+  return 0;  
 }
 
 void indexClose(SIndex *sIdx) {
@@ -61,14 +63,17 @@ void indexClose(SIndex *sIdx) {
   index_close(sIdex->index); 
   sIdx->index = NULL;
 #endif
+
+#ifdef USE_INVERTED_INDEX
   indexCacheDestroy(sIdx->cache);
-  taosHashCleanup(sIdx->fieldObj); 
+  taosHashCleanup(sIdx->colObj); 
   pthread_mutex_destroy(&sIdx->mtx);
+#endif
   free(sIdx);  
   return;
 }
 
-int indexPut(SIndex *index, SArray* fVals, int uid) {
+int indexPut(SIndex *index, SIndexMultiTerm * fVals, int uid) {
   
 #ifdef USE_LUCENE 
     index_document_t *doc = index_document_create(); 
@@ -86,32 +91,38 @@ int indexPut(SIndex *index, SArray* fVals, int uid) {
     index_document_destroy(doc);
 #endif
 
+#ifdef USE_INVERTED_INDEX
+  
    //TODO(yihao): reduce the lock range 
   pthread_mutex_lock(&index->mtx);   
   for (int i = 0; i < taosArrayGetSize(fVals); i++) {
     SIndexTerm     *p = taosArrayGetP(fVals, i);
-    SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey);
+    SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName);
     if (fi == NULL) {
-      SIdxFieldInfo tfi = {.fieldId = index->fieldId, .type = p->type};
+      SIdxColInfo tfi = {.colId = index->colId};
       index->cVersion++; 
-      index->fieldId++;
-      taosHashPut(index->fieldObj, p->key, p->nKey, &tfi, sizeof(tfi)); 
+      index->colId++;
+      taosHashPut(index->colObj, p->colName, p->nColName, &tfi, sizeof(tfi)); 
     } else {
       //TODO, del  
     }
   }   
+  pthread_mutex_unlock(&index->mtx);
 
   for (int i = 0; i < taosArrayGetSize(fVals); i++) {
     SIndexTerm     *p = taosArrayGetP(fVals, i);
-    SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey);
+    SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName);
     assert(fi !=  NULL); 
-    int32_t fieldId = fi->fieldId; 
-    int32_t colType = fi->type;  
+    int32_t colId = fi->colId; 
     int32_t version = index->cVersion;
-     
+    int ret = indexCachePut(index->cache, colId, p->colType, p->colVal, p->nColVal, version, uid, p->operType);     
+    if (ret != 0) {
+      return ret; 
+    }
   }
-  pthread_mutex_unlock(&index->mtx);
-  return 1;
+#endif
+
+  return 0;
 }
 int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result) {
 #ifdef USE_LUCENE 
@@ -148,16 +159,26 @@ int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result
   free(fields);
   free(keys);
   free(types);
+#endif
+
+#ifdef USE_INVERTED_INDEX
+  
 #endif
   return 1;
 }
 
 
 int indexDelete(SIndex *index, SIndexMultiTermQuery *query) {
+#ifdef USE_INVERTED_INDEX
+#endif
     
   return 1;
 }
-int indexRebuild(SIndex *index, SIndexOpts *opts);
+int indexRebuild(SIndex *index, SIndexOpts *opts) {
+#ifdef USE_INVERTED_INDEX
+#endif
+
+}
 
 
 SIndexOpts *indexOptsCreate() {
@@ -184,53 +205,55 @@ SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType opera) {
 void indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery) {
   for (int i = 0; i < taosArrayGetSize(pQuery->query); i++) {
     SIndexTermQuery *p = (SIndexTermQuery *)taosArrayGet(pQuery->query, i);
-    indexTermDestroy(p->field_value);
+    indexTermDestroy(p->term);
   }
   taosArrayDestroy(pQuery->query);     
   free(pQuery);
 };
-int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, const char *field, int32_t nFields, const char *value, int32_t nValue, EIndexQueryType type){
-  SIndexTerm *t = indexTermCreate(field, nFields, value, nValue);  
-  if (t == NULL) {return -1;}
-  SIndexTermQuery q = {.type = type, .field_value = t};   
+int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EIndexQueryType qType){
+  SIndexTermQuery q = {.qType = qType, .term = term};   
   taosArrayPush(pQuery->query, &q);
   return 0;
 }
 
 
-SIndexTerm *indexTermCreate(const char *key, int32_t nKey, const char *val, int32_t nVal) {
-  SIndexTerm *t = (SIndexTerm *)malloc(sizeof(SIndexTerm)); 
-  t->key  = (char *)calloc(nKey + 1, 1);
-  memcpy(t->key, key, nKey);
-  t->nKey = nKey;
+SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn oper, uint8_t colType, const char *colName, int32_t nColName, const char *colVal, int32_t nColVal) {
+  SIndexTerm *t = (SIndexTerm *)calloc(1, (sizeof(SIndexTerm))); 
+  if (t == NULL) { return NULL; }
+
+  t->suid    = suid;
+  t->operType= oper; 
+  t->colType = colType;
+
+  t->colName  = (char *)calloc(1, nColName + 1);
+  memcpy(t->colName, colName, nColName);
+  t->nColName = nColName;
 
-  t->val = (char *)calloc(nVal + 1, 1);
-  memcpy(t->val, val, nVal);
-  t->nVal = nVal;
+  t->colVal = (char *)calloc(1, nColVal + 1);
+  memcpy(t->colVal, colVal, nColVal);
+  t->nColVal = nColVal;
   return t;
 }
 void indexTermDestroy(SIndexTerm *p) {
-  free(p->key);
-  free(p->val);
+  free(p->colName);
+  free(p->colVal);
   free(p);
 }  
 
-SArray *indexMultiTermCreate() {
+SIndexMultiTerm *indexMultiTermCreate() {
   return taosArrayInit(4, sizeof(SIndexTerm *)); 
 }
 
-int indexMultiTermAdd(SArray *array, const char *field, int32_t nField, const char *val, int32_t nVal) {
-   SIndexTerm *term = indexTermCreate(field,  nField, val, nVal);  
-   if (term == NULL) { return -1; }
-   taosArrayPush(array, &term);
+int indexMultiTermAdd(SIndexMultiTerm *terms, SIndexTerm *term) {
+   taosArrayPush(terms, &term);
    return 0;
 }
-void indexMultiTermDestroy(SArray *array) {
-  for (int32_t i = 0; i < taosArrayGetSize(array); i++) {
-    SIndexTerm *p = taosArrayGetP(array, i);
+void indexMultiTermDestroy(SIndexMultiTerm *terms) {
+  for (int32_t i = 0; i < taosArrayGetSize(terms); i++) {
+    SIndexTerm *p = taosArrayGetP(terms, i);
     indexTermDestroy(p);
   }
-  taosArrayDestroy(array);
+  taosArrayDestroy(terms);
 }
 
 void indexInit() {

source/libs/index/src/index_cache.c

@@ -16,7 +16,7 @@
 #include "index_cache.h"
 #include "tcompare.h"
 
-#define MAX_INDEX_KEY_LEN 128 // test only, change later
+#define MAX_INDEX_KEY_LEN 256// test only, change later
 
 static char*  getIndexKey(const void *pData) {
   return NULL;  
@@ -96,16 +96,19 @@ IndexCache *indexCacheCreate() {
   
 }
 
-void indexCacheDestroy(IndexCache *cache) {
-  if (cache == NULL) { return; } 
-  tSkipListDestroy(cache->skiplist);
-  free(cache);
+void indexCacheDestroy(void *cache) {
+  IndexCache *pCache = cache; 
+  if (pCache == NULL) { return; } 
+  tSkipListDestroy(pCache->skiplist);
+  free(pCache);
 }
 
-int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue,  int32_t fvLen, 
+int indexCachePut(void *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue,  int32_t fvLen, 
               uint32_t version, uint64_t uid, int8_t operType) {
   if (cache == NULL) { return -1;} 
 
+  IndexCache *pCache = cache;
+
   // encode data
   int32_t total = sizeof(int32_t) + sizeof(fieldId) + sizeof(fieldType) + sizeof(fvLen) + fvLen  + sizeof(version) + sizeof(uid) + sizeof(operType); 
 
@@ -135,20 +138,15 @@ int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const c
   memcpy(p, &operType, sizeof(operType));
   p += sizeof(operType); 
 
-  tSkipListPut(cache->skiplist, (void *)buf);  
+  tSkipListPut(pCache->skiplist, (void *)buf);  
   // encode end
- 
     
 }
-int indexCacheDel(IndexCache *cache, int32_t fieldId, const char *fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) {
-
+int indexCacheDel(void *cache, int32_t fieldId, const char *fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) {
+  IndexCache *pCache = cache;
+  return 0;
 }
-int indexCacheSearch(IndexCache *cache, SIndexMultiTermQuery *query, SArray *result) {
-  
+int indexCacheSearch(void *cache, SIndexMultiTermQuery *query, SArray *result) {
+   
   return 0;  
 }
-
-
-
-
-

source/libs/index/test/CMakeLists.txt

 add_executable(indexTest "")
 target_sources(indexTest
   PRIVATE 
-  "indexTests.cpp" 
+  "indexTests.cc" 
 )
 target_include_directories ( indexTest
   PUBLIC

source/libs/index/test/indexTests.cpp → source/libs/index/test/indexTests.cc

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * 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
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
 #include <gtest/gtest.h>
 #include <string>
 #include <iostream>
@@ -61,7 +75,7 @@ class FstReadMemory {
    // add later
    bool Search(AutomationCtx *ctx, std::vector<uint64_t> &result) { 
       FstStreamBuilder *sb = fstSearch(_fst, ctx);
-      StreamWithState *st = streamBuilderIntoStream(sb);  
+      StreamWithState  *st = streamBuilderIntoStream(sb);  
       StreamWithStateResult *rt = NULL;  
       
       while ((rt = streamWithStateNextWith(st, NULL)) != NULL) {
@@ -279,15 +293,71 @@ void validateFst() {
   delete m;
 } 
 
+class IndexEnv : public ::testing::Test {
+  protected:
+    virtual void SetUp() {
+      taosRemoveDir(path);  
+      opts = indexOptsCreate(); 
+      int ret = indexOpen(opts, path, &index); 
+      assert(ret == 0);
+    } 
+    virtual void TearDown() {
+      indexClose(index);
+      indexOptsDestroy(opts);
+    }
+    
+    const char *path = "/tmp/tindex";
+    SIndexOpts *opts;  
+    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
+   {
+    
+    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);
+    }
+    {
+      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);
+   }
+   //   
+} 
 
-int main(int argc, char** argv) {
-  checkFstPerf(); 
-  //checkFstPrefixSearch();
-  return 1;
+TEST_F(IndexEnv, testDel) {
+     
 }
 
-//TEST(IndexFstBuilder, IndexFstInput) {
-//
-//}
+
+