Merge pull request #9228 from taosdata/feature/index_cache

Feature/index cache

source/libs/index/inc/index_tfile.h

@@ -59,7 +59,7 @@ typedef struct TFileReader {
 
 typedef struct IndexTFile {
   char *path;
-  TFileReader *tb;
+  TFileCache *cache;
   TFileWriter *tw;      
 } IndexTFile;
 
@@ -79,14 +79,14 @@ typedef struct TFileReaderOpt {
 } TFileReaderOpt;
 
 // tfile cache 
-TFileCache *tfileCacheCreate();
+TFileCache *tfileCacheCreate(const char *path);
 void tfileCacheDestroy(TFileCache *tcache);
 TFileReader* tfileCacheGet(TFileCache *tcache, TFileCacheKey *key);
 void tfileCachePut(TFileCache *tcache, TFileCacheKey *key, TFileReader *reader);
   
 TFileWriter *tfileWriterCreate(const char *suid, const char *colName);
 
-IndexTFile *indexTFileCreate();
+IndexTFile *indexTFileCreate(const char *path);
 
 int indexTFilePut(void *tfile, SIndexTerm *term,  uint64_t uid); 
 

source/libs/index/src/index.c

@@ -333,13 +333,17 @@ static int indexMergeFinalResults(SArray *interResults, EIndexOperatorType oType
   //refactor, merge interResults into fResults by oType
   SArray *first = taosArrayGetP(interResults, 0);  
   taosArraySort(first, uidCompare);  
+  taosArrayRemoveDuplicate(first, uidCompare, NULL);
   if (oType == MUST) {
-      
+    // just one column index, enhance later  
+    taosArrayAddAll(fResults, first); 
   } else if (oType == SHOULD) {
-     
+    // just one column index, enhance later  
+    taosArrayAddAll(fResults, first); 
     // tag1 condistion || tag2 condition
   } else if (oType == NOT) {
-    
+    // just one column index, enhance later  
+    taosArrayAddAll(fResults, first); 
     // not use currently   
   }
   return 0;

source/libs/index/src/index_cache.c

@@ -147,7 +147,7 @@ int indexCacheSearch(void *cache, SIndexTermQuery *query, int16_t colId, int32_t
 
   char *buf = calloc(1, keyLen);
   if (qtype == QUERY_TERM) {
-           
+               
   } else if (qtype == QUERY_PREFIX) {
 
   } else if (qtype == QUERY_SUFFIX) {

source/libs/index/src/index_tfile.c

@@ -13,12 +13,42 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <sys/types.h>
+#include <dirent.h>
 #include "index_tfile.h"
 #include "index_fst.h"
 #include "index_util.h"
 
 
+// tfile name suid-colId-version.tindex
+static int tfileGetFileList(const char *path, SArray *result) {
+  DIR *dir = opendir(path);  
+  if (NULL == dir) { return -1; } 
 
+  struct dirent *entry;
+  while ((entry = readdir(dir)) != NULL) {
+    size_t len = strlen(entry->d_name);
+    char *buf = calloc(1, len + 1); 
+    memcpy(buf, entry->d_name, len); 
+    taosArrayPush(result, &buf); 
+  }
+  closedir(dir);
+  return 0;
+} 
+static int tfileCompare(const void *a, const void *b) {
+  const char *aName = *(char **)a;
+  const char *bName = *(char **)b;
+  size_t aLen = strlen(aName);
+  size_t bLen = strlen(bName);
+  return strncmp(aName, bName, aLen > bLen ? aLen : bLen);
+}
+static int tfileParseFileName(const char *filename, uint64_t *suid, int *colId, int *version) {
+  if (3 == sscanf(filename, "%" PRIu64 "-%d-%d.tindex", suid, colId, version)) {
+    // read suid & colid & version  success
+    return 0;
+  }  
+  return -1;  
+} 
 static void tfileSerialCacheKey(TFileCacheKey *key, char *buf) {
   SERIALIZE_MEM_TO_BUF(buf, key, suid);
   SERIALIZE_VAR_TO_BUF(buf, '_', char); 
@@ -29,12 +59,28 @@ static void tfileSerialCacheKey(TFileCacheKey *key, char *buf) {
   SERIALIZE_STR_MEM_TO_BUF(buf, key, colName, key->nColName);
 }
 
-TFileCache *tfileCacheCreate() {
+TFileCache *tfileCacheCreate(const char *path) {
   TFileCache *tcache = calloc(1, sizeof(TFileCache)); 
   if (tcache == NULL) { return NULL; }
   
   tcache->tableCache = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); 
   tcache->capacity   = 64;
+  
+  SArray *files  = taosArrayInit(4, sizeof(void *));   
+  tfileGetFileList(path, files);   
+  taosArraySort(files, tfileCompare);
+  for (size_t i = 0; i < taosArrayGetSize(files); i++) {
+    char *file = taosArrayGetP(files, i); 
+    uint64_t suid; 
+    int colId, version;
+    if (0 != tfileParseFileName(file, &suid, &colId, &version)) {
+      // invalid file, just skip 
+      continue; 
+    } 
+    free((void *)file);    
+  }
+  taosArrayDestroy(files);
+  
   return tcache;
 }
 void tfileCacheDestroy(TFileCache *tcache) {
@@ -59,8 +105,11 @@ void tfileCachePut(TFileCache *tcache, TFileCacheKey *key, TFileReader *reader)
 
 
 
-IndexTFile *indexTFileCreate() {
+IndexTFile *indexTFileCreate(const char *path) {
   IndexTFile *tfile = calloc(1, sizeof(IndexTFile));   
+  tfile->cache = tfileCacheCreate(path);    
+  
+  
   return tfile;
 }
 void IndexTFileDestroy(IndexTFile *tfile) {
@@ -69,8 +118,15 @@ void IndexTFileDestroy(IndexTFile *tfile) {
 
 
 int indexTFileSearch(void *tfile, SIndexTermQuery *query, SArray *result) {
-  IndexTFile *ptfile = (IndexTFile *)tfile;
-     
+  IndexTFile *pTfile = (IndexTFile *)tfile;
+  
+  SIndexTerm *term = query->term;
+  TFileCacheKey key = {.suid    = term->suid,
+                       .colType = term->colType,
+                       .version = 0,
+                       .colName = term->colName, 
+                       .nColName= term->nColName};
+  TFileReader *reader = tfileCacheGet(pTfile->cache, &key);    
   return 0;
 }
 int indexTFilePut(void *tfile, SIndexTerm *term,  uint64_t uid) {

source/libs/index/test/indexTests.cc

@@ -26,6 +26,7 @@
 class FstWriter {
   public:
     FstWriter() {
+      _wc = writerCtxCreate(TFile, "/tmp/tindex", false, 0); 
       _b = fstBuilderCreate(NULL, 0);
     }  
    bool Put(const std::string &key, uint64_t val) {
@@ -37,15 +38,19 @@ class FstWriter {
    ~FstWriter() {
      fstBuilderFinish(_b);
      fstBuilderDestroy(_b);
+
+     writerCtxDestroy(_wc);
    }
   private:
     FstBuilder *_b; 
+    WriterCtx *_wc;
 };
 
 class FstReadMemory {
   public:
    FstReadMemory(size_t size) {
-     _w    = fstCountingWriterCreate(NULL); 
+     _wc   = writerCtxCreate(TFile, "/tmp/tindex", true, 0);   
+     _w    = fstCountingWriterCreate(_wc); 
      _size = size; 
      memset((void *)&_s, 0, sizeof(_s));
    }
@@ -94,12 +99,14 @@ class FstReadMemory {
     fstCountingWriterDestroy(_w);
     fstDestroy(_fst);
     fstSliceDestroy(&_s);
+    writerCtxDestroy(_wc);
   } 
   
   private:
    FstCountingWriter *_w; 
    Fst *_fst;
    FstSlice _s;  
+   WriterCtx *_wc;
    size_t _size;
    
 };