Merge remote-tracking branch 'origin/3.0' into feature/dnode3

source/libs/index/inc/index_fst_counting_writer.h

@@ -37,6 +37,7 @@ typedef struct WriterCtx {
     struct {
       int  fd;
       bool readOnly;
+      char buf[256];
     } file;
     struct {
       int32_t capa;
@@ -53,7 +54,7 @@ static int writeCtxDoReadFrom(WriterCtx* ctx, uint8_t* buf, int len, int32_t off
 static int writeCtxDoFlush(WriterCtx* ctx);
 
 WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int32_t capacity);
-void       writerCtxDestroy(WriterCtx* w);
+void       writerCtxDestroy(WriterCtx* w, bool remove);
 
 typedef uint32_t CheckSummer;
 

source/libs/index/inc/index_tfile.h

@@ -107,6 +107,7 @@ void         tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader*
 
 TFileReader* tfileGetReaderByCol(IndexTFile* tf, char* colName);
 
+TFileReader* tfileReaderOpen(char* path, uint64_t suid, int32_t version, const char* colName);
 TFileReader* tfileReaderCreate(WriterCtx* ctx);
 void         tfileReaderDestroy(TFileReader* reader);
 int          tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* result);
@@ -114,10 +115,10 @@ void         tfileReaderRef(TFileReader* reader);
 void         tfileReaderUnRef(TFileReader* reader);
 
 TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int32_t version, const char* colName, uint8_t type);
-void         tfileWriteClose(TFileWriter* tw);
+void         tfileWriterClose(TFileWriter* tw);
 TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header);
 void         tfileWriterDestroy(TFileWriter* tw);
-int          tfileWriterPut(TFileWriter* tw, void* data);
+int          tfileWriterPut(TFileWriter* tw, void* data, bool order);
 int          tfileWriterFinish(TFileWriter* tw);
 
 //

source/libs/index/src/index.c

@@ -25,7 +25,7 @@
 #endif
 
 #define INDEX_NUM_OF_THREADS 4
-#define INDEX_QUEUE_SIZE 4
+#define INDEX_QUEUE_SIZE 200
 
 void* indexQhandle = NULL;
 
@@ -52,18 +52,17 @@ typedef struct SIdxColInfo {
 
 static pthread_once_t isInit = PTHREAD_ONCE_INIT;
 // static void           indexInit();
-
 static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* term, SArray** result);
 
 static void indexInterResultsDestroy(SArray* results);
 static int  indexMergeFinalResults(SArray* interResults, EIndexOperatorType oType, SArray* finalResult);
 
+static int indexGenTFile(SIndex* index, IndexCache* cache, SArray* batch);
+
 int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
   // pthread_once(&isInit, indexInit);
   SIndex* sIdx = calloc(1, sizeof(SIndex));
-  if (sIdx == NULL) {
-    return -1;
-  }
+  if (sIdx == NULL) { return -1; }
 
 #ifdef USE_LUCENE
   index_t* index = index_open(path);
@@ -99,9 +98,7 @@ void indexClose(SIndex* sIdx) {
   void* iter = taosHashIterate(sIdx->colObj, NULL);
   while (iter) {
     IndexCache** pCache = iter;
-    if (*pCache) {
-      indexCacheUnRef(*pCache);
-    }
+    if (*pCache) { indexCacheUnRef(*pCache); }
     iter = taosHashIterate(sIdx->colObj, iter);
   }
   taosHashCleanup(sIdx->colObj);
@@ -147,9 +144,7 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
     IndexCache** cache = taosHashGet(index->colObj, p->colName, p->nColName);
     assert(*cache != NULL);
     int ret = indexCachePut(*cache, p, uid);
-    if (ret != 0) {
-      return ret;
-    }
+    if (ret != 0) { return ret; }
   }
 
 #endif
@@ -179,9 +174,7 @@ int indexSearch(SIndex* index, SIndexMultiTermQuery* multiQuerys, SArray* result
   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 < tsz; i++) { taosArrayPush(result, &tResult[i]); }
 
   for (int i = 0; i < nQuery; i++) {
     free(fields[i]);
@@ -238,9 +231,7 @@ void indexOptsDestroy(SIndexOpts* opts) {
  */
 SIndexMultiTermQuery* indexMultiTermQueryCreate(EIndexOperatorType opera) {
   SIndexMultiTermQuery* p = (SIndexMultiTermQuery*)malloc(sizeof(SIndexMultiTermQuery));
-  if (p == NULL) {
-    return NULL;
-  }
+  if (p == NULL) { return NULL; }
   p->opera = opera;
   p->query = taosArrayInit(4, sizeof(SIndexTermQuery));
   return p;
@@ -262,9 +253,7 @@ int indexMultiTermQueryAdd(SIndexMultiTermQuery* pQuery, SIndexTerm* term, EInde
 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;
-  }
+  if (t == NULL) { return NULL; }
 
   t->suid = suid;
   t->operType = oper;
@@ -317,7 +306,7 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result
 
   *result = taosArrayInit(4, sizeof(uint64_t));
   // TODO: iterator mem and tidex
-  STermValueType s;
+  STermValueType s = kTypeValue;
   if (0 == indexCacheSearch(cache, query, *result, &s)) {
     if (s == kTypeDeletion) {
       indexInfo("col: %s already drop by other opera", term->colName);
@@ -336,9 +325,7 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result
   return 0;
 }
 static void indexInterResultsDestroy(SArray* results) {
-  if (results == NULL) {
-    return;
-  }
+  if (results == NULL) { return; }
 
   size_t sz = taosArrayGetSize(results);
   for (size_t i = 0; i < sz; i++) {
@@ -368,23 +355,43 @@ static int indexMergeFinalResults(SArray* interResults, EIndexOperatorType oType
   return 0;
 }
 
-int indexFlushCacheTFile(SIndex* sIdx, void* cache) {
-  if (sIdx == NULL) {
-    return -1;
+static void indexMergeSameKey(SArray* result, TFileValue* tv) {
+  int32_t sz = result ? taosArrayGetSize(result) : 0;
+  if (sz > 0) {
+    // TODO(yihao): remove duplicate tableid
+    TFileValue* lv = taosArrayGetP(result, sz - 1);
+    if (strcmp(lv->colVal, tv->colVal) == 0) {
+      taosArrayAddAll(lv->tableId, tv->tableId);
+      tfileValueDestroy(tv);
+    } else {
+      taosArrayPush(result, &tv);
+    }
+  } else {
+    taosArrayPush(result, &tv);
   }
+}
+static void indexDestroyTempResult(SArray* result) {
+  int32_t sz = result ? taosArrayGetSize(result) : 0;
+  for (size_t i = 0; i < sz; i++) {
+    TFileValue* tv = taosArrayGetP(result, i);
+    tfileValueDestroy(tv);
+  }
+  taosArrayDestroy(result);
+}
+int indexFlushCacheTFile(SIndex* sIdx, void* cache) {
+  if (sIdx == NULL) { return -1; }
   indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid);
 
   IndexCache*  pCache = (IndexCache*)cache;
   TFileReader* pReader = tfileGetReaderByCol(sIdx->tindex, pCache->colName);
-
   // handle flush
   Iterate* cacheIter = indexCacheIteratorCreate(pCache);
   Iterate* tfileIter = tfileIteratorCreate(pReader);
 
   SArray* result = taosArrayInit(1024, sizeof(void*));
 
-  bool cn = cacheIter->next(cacheIter);
-  bool tn = tfileIter->next(tfileIter);
+  bool cn = cacheIter ? cacheIter->next(cacheIter) : false;
+  bool tn = tfileIter ? tfileIter->next(tfileIter) : false;
   while (cn == true && tn == true) {
     IterateValue* cv = cacheIter->getValue(cacheIter);
     IterateValue* tv = tfileIter->getValue(tfileIter);
@@ -395,7 +402,7 @@ int indexFlushCacheTFile(SIndex* sIdx, void* cache) {
       TFileValue* tfv = tfileValueCreate(cv->colVal);
       taosArrayAddAll(tfv->tableId, cv->val);
       taosArrayAddAll(tfv->tableId, tv->val);
-      taosArrayPush(result, &tfv);
+      indexMergeSameKey(result, tfv);
 
       cn = cacheIter->next(cacheIter);
       tn = tfileIter->next(tfileIter);
@@ -403,13 +410,15 @@ int indexFlushCacheTFile(SIndex* sIdx, void* cache) {
     } else if (comp < 0) {
       TFileValue* tfv = tfileValueCreate(cv->colVal);
       taosArrayAddAll(tfv->tableId, cv->val);
-      taosArrayPush(result, &tfv);
+
+      indexMergeSameKey(result, tfv);
       // copy to final Result;
       cn = cacheIter->next(cacheIter);
     } else {
       TFileValue* tfv = tfileValueCreate(tv->colVal);
-      taosArrayPush(result, &tfv);
       taosArrayAddAll(tfv->tableId, tv->val);
+
+      indexMergeSameKey(result, tfv);
       // copy to final result
       tn = tfileIter->next(tfileIter);
     }
@@ -418,33 +427,24 @@ int indexFlushCacheTFile(SIndex* sIdx, void* cache) {
     IterateValue* cv = cacheIter->getValue(cacheIter);
     TFileValue*   tfv = tfileValueCreate(cv->colVal);
     taosArrayAddAll(tfv->tableId, cv->val);
-    taosArrayPush(result, &tfv);
+    indexMergeSameKey(result, tfv);
     cn = cacheIter->next(cacheIter);
   }
   while (tn == true) {
     IterateValue* tv = tfileIter->getValue(tfileIter);
     TFileValue*   tfv = tfileValueCreate(tv->colVal);
+    if (tv->val == NULL) {
+      // HO
+      printf("NO....");
+    }
     taosArrayAddAll(tfv->tableId, tv->val);
-    taosArrayPush(result, &tfv);
+    indexMergeSameKey(result, tfv);
     tn = tfileIter->next(tfileIter);
   }
-
-  int32_t version = CACHE_VERSION(pCache);
-  uint8_t colType = pCache->type;
-
-  TFileWriter* tw = tfileWriterOpen(sIdx->path, sIdx->suid, version, pCache->colName, colType);
-  if (tw == NULL) {
-    indexError("faile to open file to write");
-  } else {
-    int ret = tfileWriterPut(tw, result);
-    if (ret != 0) {
-      indexError("faile to write into tindex ");
-    }
-  }
-  // not free later, just put int table cache
+  int ret = indexGenTFile(sIdx, pCache, result);
+  indexDestroyTempResult(result);
   indexCacheDestroyImm(pCache);
 
-  tfileWriteClose(tw);
   indexCacheIteratorDestroy(cacheIter);
   tfileIteratorDestroy(tfileIter);
 
@@ -455,9 +455,45 @@ int indexFlushCacheTFile(SIndex* sIdx, void* cache) {
 void iterateValueDestroy(IterateValue* value, bool destroy) {
   if (destroy) {
     taosArrayDestroy(value->val);
+    value->val = NULL;
   } else {
-    taosArrayClear(value->val);
+    if (value->val != NULL) { taosArrayClear(value->val); }
   }
-  free(value->colVal);
+  // free(value->colVal);
   value->colVal = NULL;
 }
+static int indexGenTFile(SIndex* sIdx, IndexCache* cache, SArray* batch) {
+  int32_t version = CACHE_VERSION(cache);
+  uint8_t colType = cache->type;
+
+  TFileWriter* tw = tfileWriterOpen(sIdx->path, sIdx->suid, version, cache->colName, colType);
+  if (tw == NULL) {
+    indexError("failed to open file to write");
+    return -1;
+  }
+
+  int ret = tfileWriterPut(tw, batch, true);
+  if (ret != 0) {
+    indexError("failed to write into tindex ");
+    goto END;
+  }
+  tfileWriterClose(tw);
+
+  TFileReader* reader = tfileReaderOpen(sIdx->path, sIdx->suid, version, cache->colName);
+
+  char          buf[128] = {0};
+  TFileHeader*  header = &reader->header;
+  TFileCacheKey key = {.suid = header->suid,
+                       .colName = header->colName,
+                       .nColName = strlen(header->colName),
+                       .colType = header->colType};
+  pthread_mutex_lock(&sIdx->mtx);
+
+  IndexTFile* ifile = (IndexTFile*)sIdx->tindex;
+  tfileCachePut(ifile->cache, &key, reader);
+
+  pthread_mutex_unlock(&sIdx->mtx);
+  return ret;
+END:
+  tfileWriterClose(tw);
+}

source/libs/index/src/index_cache.c

@@ -20,7 +20,7 @@
 
 #define MAX_INDEX_KEY_LEN 256  // test only, change later
 
-#define MEM_TERM_LIMIT 1000000
+#define MEM_TERM_LIMIT 100
 // 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) +
@@ -86,25 +86,26 @@ void indexCacheDestroySkiplist(SSkipList* slt) {
   while (tSkipListIterNext(iter)) {
     SSkipListNode* node = tSkipListIterGet(iter);
     CacheTerm*     ct = (CacheTerm*)SL_GET_NODE_DATA(node);
-    if (ct != NULL) {
-    }
+    if (ct != NULL) {}
   }
   tSkipListDestroyIter(iter);
   tSkipListDestroy(slt);
 }
 void indexCacheDestroyImm(IndexCache* cache) {
+  if (cache == NULL) { return; }
+
   MemTable* tbl = NULL;
   pthread_mutex_lock(&cache->mtx);
   tbl = cache->imm;
   cache->imm = NULL;  // or throw int bg thread
   pthread_mutex_unlock(&cache->mtx);
+
+  indexMemUnRef(tbl);
   indexMemUnRef(tbl);
 }
 void indexCacheDestroy(void* cache) {
   IndexCache* pCache = cache;
-  if (pCache == NULL) {
-    return;
-  }
+  if (pCache == NULL) { return; }
   indexMemUnRef(pCache->mem);
   indexMemUnRef(pCache->imm);
   free(pCache->colName);
@@ -114,9 +115,11 @@ void indexCacheDestroy(void* cache) {
 
 Iterate* indexCacheIteratorCreate(IndexCache* cache) {
   Iterate* iiter = calloc(1, sizeof(Iterate));
-  if (iiter == NULL) {
-    return NULL;
-  }
+  if (iiter == NULL) { return NULL; }
+
+  pthread_mutex_lock(&cache->mtx);
+
+  indexMemRef(cache->imm);
 
   MemTable* tbl = cache->imm;
   iiter->val.val = taosArrayInit(1, sizeof(uint64_t));
@@ -124,12 +127,12 @@ Iterate* indexCacheIteratorCreate(IndexCache* cache) {
   iiter->next = indexCacheIteratorNext;
   iiter->getValue = indexCacheIteratorGetValue;
 
+  pthread_mutex_unlock(&cache->mtx);
+
   return iiter;
 }
 void indexCacheIteratorDestroy(Iterate* iter) {
-  if (iter == NULL) {
-    return;
-  }
+  if (iter == NULL) { return; }
   tSkipListDestroyIter(iter->iter);
   iterateValueDestroy(&iter->val, true);
   free(iter);
@@ -155,6 +158,7 @@ static void indexCacheMakeRoomForWrite(IndexCache* cache) {
       taosMsleep(50);
       pthread_mutex_lock(&cache->mtx);
     } else {
+      indexCacheRef(cache);
       cache->imm = cache->mem;
       cache->mem = indexInternalCacheCreate(cache->type);
       cache->nTerm = 1;
@@ -166,17 +170,13 @@ static void indexCacheMakeRoomForWrite(IndexCache* cache) {
 }
 
 int indexCachePut(void* cache, SIndexTerm* term, uint64_t uid) {
-  if (cache == NULL) {
-    return -1;
-  }
+  if (cache == NULL) { return -1; }
 
   IndexCache* pCache = cache;
   indexCacheRef(pCache);
   // encode data
   CacheTerm* ct = calloc(1, sizeof(CacheTerm));
-  if (cache == NULL) {
-    return -1;
-  }
+  if (cache == NULL) { return -1; }
   // set up key
   ct->colType = term->colType;
   ct->colVal = (char*)calloc(1, sizeof(char) * (term->nColVal + 1));
@@ -205,32 +205,11 @@ int indexCacheDel(void* cache, const char* fieldValue, int32_t fvlen, uint64_t u
   IndexCache* pCache = cache;
   return 0;
 }
-int indexCacheSearch(void* cache, SIndexTermQuery* query, SArray* result, STermValueType* s) {
-  if (cache == NULL) {
-    return -1;
-  }
-  IndexCache*     pCache = cache;
-  SIndexTerm*     term = query->term;
-  EIndexQueryType qtype = query->qType;
-
-  MemTable *mem = NULL, *imm = NULL;
-  pthread_mutex_lock(&pCache->mtx);
-  mem = pCache->mem;
-  imm = pCache->imm;
-  indexMemRef(mem);
-  indexMemRef(imm);
-  pthread_mutex_unlock(&pCache->mtx);
-
-  CacheTerm* ct = calloc(1, sizeof(CacheTerm));
-  if (ct == NULL) {
-    return -1;
-  }
-  ct->colVal = calloc(1, sizeof(char) * (term->nColVal + 1));
-  memcpy(ct->colVal, term->colVal, term->nColVal);
-  ct->version = atomic_load_32(&pCache->version);
 
+static int indexQueryMem(MemTable* mem, CacheTerm* ct, EIndexQueryType qtype, SArray* result, STermValueType* s) {
+  if (mem == NULL) { return 0; }
   char* key = getIndexKey(ct);
-  // TODO handle multi situation later, and refactor
+
   SSkipListIterator* iter = tSkipListCreateIterFromVal(mem->mem, key, TSDB_DATA_TYPE_BINARY, TSDB_ORDER_ASC);
   while (tSkipListIterNext(iter)) {
     SSkipListNode* node = tSkipListIterGet(iter);
@@ -251,51 +230,55 @@ int indexCacheSearch(void* cache, SIndexTermQuery* query, SArray* result, STermV
     }
   }
   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) {
-    //
-  } else if (qtype == QUERY_SUFFIX) {
-    //
-  } else if (qtype == QUERY_REGEX) {
-    //
+  return 0;
+}
+int indexCacheSearch(void* cache, SIndexTermQuery* query, SArray* result, STermValueType* s) {
+  if (cache == NULL) { return -1; }
+  IndexCache* pCache = cache;
+
+  MemTable *mem = NULL, *imm = NULL;
+  pthread_mutex_lock(&pCache->mtx);
+  mem = pCache->mem;
+  imm = pCache->imm;
+  indexMemRef(mem);
+  indexMemRef(imm);
+  pthread_mutex_unlock(&pCache->mtx);
+
+  SIndexTerm*     term = query->term;
+  EIndexQueryType qtype = query->qType;
+  CacheTerm       ct = {.colVal = term->colVal, .version = atomic_load_32(&pCache->version)};
+
+  int ret = indexQueryMem(mem, &ct, qtype, result, s);
+  if (ret == 0 && *s != kTypeDeletion) {
+    // continue search in imm
+    ret = indexQueryMem(imm, &ct, qtype, result, s);
   }
+  // cacheTermDestroy(ct);
+
   indexMemUnRef(mem);
   indexMemUnRef(imm);
-  return 0;
+
+  return ret;
 }
 
 void indexCacheRef(IndexCache* cache) {
-  if (cache == NULL) {
-    return;
-  }
+  if (cache == NULL) { return; }
   int ref = T_REF_INC(cache);
   UNUSED(ref);
 }
 void indexCacheUnRef(IndexCache* cache) {
-  if (cache == NULL) {
-    return;
-  }
+  if (cache == NULL) { return; }
   int ref = T_REF_DEC(cache);
-  if (ref == 0) {
-    indexCacheDestroy(cache);
-  }
+  if (ref == 0) { indexCacheDestroy(cache); }
 }
 
 void indexMemRef(MemTable* tbl) {
-  if (tbl == NULL) {
-    return;
-  }
+  if (tbl == NULL) { return; }
   int ref = T_REF_INC(tbl);
   UNUSED(ref);
 }
 void indexMemUnRef(MemTable* tbl) {
-  if (tbl == NULL) {
-    return;
-  }
+  if (tbl == NULL) { return; }
   int ref = T_REF_DEC(tbl);
   if (ref == 0) {
     SSkipList* slt = tbl->mem;
@@ -305,9 +288,7 @@ void indexMemUnRef(MemTable* tbl) {
 }
 
 static void cacheTermDestroy(CacheTerm* ct) {
-  if (ct == NULL) {
-    return;
-  }
+  if (ct == NULL) { return; }
   free(ct->colVal);
   free(ct);
 }
@@ -322,9 +303,7 @@ static int32_t compareKey(const void* l, const void* r) {
 
   // compare colVal
   int32_t cmp = strcmp(lt->colVal, rt->colVal);
-  if (cmp == 0) {
-    return rt->version - lt->version;
-  }
+  if (cmp == 0) { return rt->version - lt->version; }
   return cmp;
 }
 
@@ -344,9 +323,7 @@ static void doMergeWork(SSchedMsg* msg) {
 }
 static bool indexCacheIteratorNext(Iterate* itera) {
   SSkipListIterator* iter = itera->iter;
-  if (iter == NULL) {
-    return false;
-  }
+  if (iter == NULL) { return false; }
   IterateValue* iv = &itera->val;
   iterateValueDestroy(iv, false);
 
@@ -356,7 +333,8 @@ static bool indexCacheIteratorNext(Iterate* itera) {
     CacheTerm*     ct = (CacheTerm*)SL_GET_NODE_DATA(node);
 
     iv->type = ct->operaType;
-    iv->colVal = ct->colVal;
+    iv->colVal = calloc(1, strlen(ct->colVal) + 1);
+    memcpy(iv->colVal, ct->colVal, strlen(ct->colVal));
 
     taosArrayPush(iv->val, &ct->uid);
   }

source/libs/index/src/index_fst_automation.c

@@ -17,7 +17,9 @@
 
 StartWithStateValue* startWithStateValueCreate(StartWithStateKind kind, ValueType ty, void* val) {
   StartWithStateValue* nsv = calloc(1, sizeof(StartWithStateValue));
-  if (nsv == NULL) { return NULL; }
+  if (nsv == NULL) {
+    return NULL;
+  }
 
   nsv->kind = kind;
   nsv->type = ty;
@@ -35,7 +37,9 @@ StartWithStateValue* startWithStateValueCreate(StartWithStateKind kind, ValueTyp
 }
 void startWithStateValueDestroy(void* val) {
   StartWithStateValue* sv = (StartWithStateValue*)val;
-  if (sv == NULL) { return; }
+  if (sv == NULL) {
+    return;
+  }
 
   if (sv->type == FST_INT) {
     //
@@ -48,7 +52,9 @@ void startWithStateValueDestroy(void* val) {
 }
 StartWithStateValue* startWithStateValueDump(StartWithStateValue* sv) {
   StartWithStateValue* nsv = calloc(1, sizeof(StartWithStateValue));
-  if (nsv == NULL) { return NULL; }
+  if (nsv == NULL) {
+    return NULL;
+  }
 
   nsv->kind = sv->kind;
   nsv->type = sv->type;
@@ -65,24 +71,12 @@ StartWithStateValue* startWithStateValueDump(StartWithStateValue* sv) {
 }
 
 // iterate fst
-static void* alwaysMatchStart(AutomationCtx* ctx) {
-  return NULL;
-}
-static bool alwaysMatchIsMatch(AutomationCtx* ctx, void* state) {
-  return true;
-}
-static bool alwaysMatchCanMatch(AutomationCtx* ctx, void* state) {
-  return true;
-}
-static bool alwaysMatchWillAlwaysMatch(AutomationCtx* ctx, void* state) {
-  return true;
-}
-static void* alwaysMatchAccpet(AutomationCtx* ctx, void* state, uint8_t byte) {
-  return NULL;
-}
-static void* alwaysMatchAccpetEof(AutomationCtx* ctx, void* state) {
-  return NULL;
-}
+static void* alwaysMatchStart(AutomationCtx* ctx) { return NULL; }
+static bool  alwaysMatchIsMatch(AutomationCtx* ctx, void* state) { return true; }
+static bool  alwaysMatchCanMatch(AutomationCtx* ctx, void* state) { return true; }
+static bool  alwaysMatchWillAlwaysMatch(AutomationCtx* ctx, void* state) { return true; }
+static void* alwaysMatchAccpet(AutomationCtx* ctx, void* state, uint8_t byte) { return NULL; }
+static void* alwaysMatchAccpetEof(AutomationCtx* ctx, void* state) { return NULL; }
 // prefix query, impl later
 
 static void* prefixStart(AutomationCtx* ctx) {
@@ -97,17 +91,20 @@ static bool prefixCanMatch(AutomationCtx* ctx, void* sv) {
   StartWithStateValue* ssv = (StartWithStateValue*)sv;
   return ssv->val >= 0;
 }
-static bool prefixWillAlwaysMatch(AutomationCtx* ctx, void* state) {
-  return true;
-}
+static bool  prefixWillAlwaysMatch(AutomationCtx* ctx, void* state) { return true; }
 static void* prefixAccept(AutomationCtx* ctx, void* state, uint8_t byte) {
   StartWithStateValue* ssv = (StartWithStateValue*)state;
-  if (ssv == NULL || ctx == NULL) { return NULL; }
+  if (ssv == NULL || ctx == NULL) {
+    return NULL;
+  }
 
   char* data = ctx->data;
-  if (ssv->kind == Done) { return startWithStateValueCreate(Done, FST_INT, &ssv->val); }
+  if (ssv->kind == Done) {
+    return startWithStateValueCreate(Done, FST_INT, &ssv->val);
+  }
   if ((strlen(data) > ssv->val) && data[ssv->val] == byte) {
-    int                  val = ssv->val + 1;
+    int val = ssv->val + 1;
+
     StartWithStateValue* nsv = startWithStateValueCreate(Running, FST_INT, &val);
     if (prefixIsMatch(ctx, nsv)) {
       nsv->kind = Done;
@@ -118,35 +115,22 @@ static void* prefixAccept(AutomationCtx* ctx, void* state, uint8_t byte) {
   }
   return NULL;
 }
-static void* prefixAcceptEof(AutomationCtx* ctx, void* state) {
-  return NULL;
-}
+static void* prefixAcceptEof(AutomationCtx* ctx, void* state) { return NULL; }
 
 // pattern query, impl later
 
-static void* patternStart(AutomationCtx* ctx) {
-  return NULL;
-}
-static bool patternIsMatch(AutomationCtx* ctx, void* data) {
-  return true;
-}
-static bool patternCanMatch(AutomationCtx* ctx, void* data) {
-  return true;
-}
-static bool patternWillAlwaysMatch(AutomationCtx* ctx, void* state) {
-  return true;
-}
+static void* patternStart(AutomationCtx* ctx) { return NULL; }
+static bool  patternIsMatch(AutomationCtx* ctx, void* data) { return true; }
+static bool  patternCanMatch(AutomationCtx* ctx, void* data) { return true; }
+static bool  patternWillAlwaysMatch(AutomationCtx* ctx, void* state) { return true; }
 
-static void* patternAccept(AutomationCtx* ctx, void* state, uint8_t byte) {
-  return NULL;
-}
+static void* patternAccept(AutomationCtx* ctx, void* state, uint8_t byte) { return NULL; }
 
-static void* patternAcceptEof(AutomationCtx* ctx, void* state) {
-  return NULL;
-}
+static void* patternAcceptEof(AutomationCtx* ctx, void* state) { return NULL; }
 
 AutomationFunc automFuncs[] = {
-    {alwaysMatchStart, alwaysMatchIsMatch, alwaysMatchCanMatch, alwaysMatchWillAlwaysMatch, alwaysMatchAccpet, alwaysMatchAccpetEof},
+    {alwaysMatchStart, alwaysMatchIsMatch, alwaysMatchCanMatch, alwaysMatchWillAlwaysMatch, alwaysMatchAccpet,
+     alwaysMatchAccpetEof},
     {prefixStart, prefixIsMatch, prefixCanMatch, prefixWillAlwaysMatch, prefixAccept, prefixAcceptEof},
     {patternStart, patternIsMatch, patternCanMatch, patternWillAlwaysMatch, patternAccept, patternAcceptEof}
     // add more search type
@@ -154,7 +138,9 @@ AutomationFunc automFuncs[] = {
 
 AutomationCtx* automCtxCreate(void* data, AutomationType atype) {
   AutomationCtx* ctx = calloc(1, sizeof(AutomationCtx));
-  if (ctx == NULL) { return NULL; }
+  if (ctx == NULL) {
+    return NULL;
+  }
 
   StartWithStateValue* sv = NULL;
   if (atype == AUTOMATION_ALWAYS) {
@@ -170,11 +156,14 @@ AutomationCtx* automCtxCreate(void* data, AutomationType atype) {
     // add more search type
   }
 
-  char*  src = (char*)data;
-  size_t len = strlen(src);
-  char*  dst = (char*)malloc(len * sizeof(char) + 1);
-  memcpy(dst, src, len);
-  dst[len] = 0;
+  char* dst = NULL;
+  if (data != NULL) {
+    char*  src = (char*)data;
+    size_t len = strlen(src);
+    dst = (char*)malloc(len * sizeof(char) + 1);
+    memcpy(dst, src, len);
+    dst[len] = 0;
+  }
 
   ctx->data = dst;
   ctx->type = atype;

source/libs/index/src/index_fst_counting_writer.c

@@ -73,6 +73,7 @@ WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int
     } else {
       ctx->file.fd = tfOpenReadWrite(path);
     }
+    memcpy(ctx->file.buf, path, strlen(path));
     if (ctx->file.fd < 0) {
       indexError("open file error %d", errno);
       goto END;
@@ -95,11 +96,12 @@ END:
   free(ctx);
   return NULL;
 }
-void writerCtxDestroy(WriterCtx* ctx) {
+void writerCtxDestroy(WriterCtx* ctx, bool remove) {
   if (ctx->type == TMemory) {
     free(ctx->mem.buf);
   } else {
     tfClose(ctx->file.fd);
+    if (remove) unlink(ctx->file.buf);
   }
   free(ctx);
 }
@@ -138,10 +140,8 @@ int fstCountingWriterRead(FstCountingWriter* write, uint8_t* buf, uint32_t len)
   return nRead;
 }
 
-uint32_t fstCountingWriterMaskedCheckSum(FstCountingWriter* write) {
-  return 0;
-}
-int fstCountingWriterFlush(FstCountingWriter* write) {
+uint32_t fstCountingWriterMaskedCheckSum(FstCountingWriter* write) { return 0; }
+int      fstCountingWriterFlush(FstCountingWriter* write) {
   WriterCtx* ctx = write->wrt;
   ctx->flush(ctx);
   // write->wtr->flush

source/libs/index/src/index_tfile.c

@@ -32,6 +32,7 @@ typedef struct TFileFstIter {
 
 #define TF_TABLE_TATOAL_SIZE(sz) (sizeof(sz) + sz * sizeof(uint64_t))
 
+static int  tfileUidCompare(const void* a, const void* b);
 static int  tfileStrCompare(const void* a, const void* b);
 static int  tfileValueCompare(const void* a, const void* b, const void* param);
 static void tfileSerialTableIdsToBuf(char* buf, SArray* tableIds);
@@ -52,11 +53,16 @@ static int  tfileParseFileName(const char* filename, uint64_t* suid, int* colId,
 static void tfileGenFileName(char* filename, uint64_t suid, int colId, int version);
 static void tfileSerialCacheKey(TFileCacheKey* key, char* buf);
 
+static TFileReader* tfileReaderCreateImpl(WriterCtx* ctx) {
+  TFileReader* reader = tfileReaderCreate(ctx);
+  tfileReaderRef(reader);
+  // tfileSerialCacheKey(&key, buf);
+  return reader;
+}
+
 TFileCache* tfileCacheCreate(const char* path) {
   TFileCache* tcache = calloc(1, sizeof(TFileCache));
-  if (tcache == NULL) {
-    return NULL;
-  }
+  if (tcache == NULL) { return NULL; }
 
   tcache->tableCache = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
   tcache->capacity = 64;
@@ -81,16 +87,13 @@ TFileCache* tfileCacheCreate(const char* path) {
       goto End;
     }
 
-    TFileReader* reader = tfileReaderCreate(wc);
-    tfileReaderRef(reader);
-    // loader fst and validate it
+    char          buf[128] = {0};
+    TFileReader*  reader = tfileReaderCreateImpl(wc);
     TFileHeader*  header = &reader->header;
     TFileCacheKey key = {.suid = header->suid,
                          .colName = header->colName,
                          .nColName = strlen(header->colName),
                          .colType = header->colType};
-
-    char buf[128] = {0};
     tfileSerialCacheKey(&key, buf);
     taosHashPut(tcache->tableCache, buf, strlen(buf), &reader, sizeof(void*));
   }
@@ -102,9 +105,7 @@ End:
   return NULL;
 }
 void tfileCacheDestroy(TFileCache* tcache) {
-  if (tcache == NULL) {
-    return;
-  }
+  if (tcache == NULL) { return; }
 
   // free table cache
   TFileReader** reader = taosHashIterate(tcache->tableCache, NULL);
@@ -125,9 +126,7 @@ TFileReader* tfileCacheGet(TFileCache* tcache, TFileCacheKey* key) {
   tfileSerialCacheKey(key, buf);
 
   TFileReader** reader = taosHashGet(tcache->tableCache, buf, strlen(buf));
-  if (reader == NULL) {
-    return NULL;
-  }
+  if (reader == NULL) { return NULL; }
   tfileReaderRef(*reader);
 
   return *reader;
@@ -137,7 +136,7 @@ void tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader* reader)
   tfileSerialCacheKey(key, buf);
   // remove last version index reader
   TFileReader** p = taosHashGet(tcache->tableCache, buf, strlen(buf));
-  if (*p != NULL) {
+  if (p != NULL) {
     TFileReader* oldReader = *p;
     taosHashRemove(tcache->tableCache, buf, strlen(buf));
     tfileReaderUnRef(oldReader);
@@ -149,9 +148,7 @@ void tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader* reader)
 }
 TFileReader* tfileReaderCreate(WriterCtx* ctx) {
   TFileReader* reader = calloc(1, sizeof(TFileReader));
-  if (reader == NULL) {
-    return NULL;
-  }
+  if (reader == NULL) { return NULL; }
 
   // T_REF_INC(reader);
   reader->ctx = ctx;
@@ -172,12 +169,10 @@ TFileReader* tfileReaderCreate(WriterCtx* ctx) {
   return reader;
 }
 void tfileReaderDestroy(TFileReader* reader) {
-  if (reader == NULL) {
-    return;
-  }
+  if (reader == NULL) { return; }
   // T_REF_INC(reader);
   fstDestroy(reader->fst);
-  writerCtxDestroy(reader->ctx);
+  writerCtxDestroy(reader->ctx, true);
   free(reader);
 }
 
@@ -216,7 +211,8 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int32_t version, const c
 
   char fullname[256] = {0};
   snprintf(fullname, sizeof(fullname), "%s/%s", path, filename);
-  WriterCtx* wcx = writerCtxCreate(TFile, fullname, true, 1024 * 1024);
+  WriterCtx* wcx = writerCtxCreate(TFile, fullname, false, 1024 * 1024);
+  if (wcx == NULL) { return NULL; }
 
   TFileHeader tfh = {0};
   tfh.suid = suid;
@@ -226,6 +222,21 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int32_t version, const c
 
   return tfileWriterCreate(wcx, &tfh);
 }
+TFileReader* tfileReaderOpen(char* path, uint64_t suid, int32_t version, const char* colName) {
+  char    filename[128] = {0};
+  int32_t coldId = 1;
+  tfileGenFileName(filename, suid, coldId, version);
+
+  char fullname[256] = {0};
+  snprintf(fullname, sizeof(fullname), "%s/%s", path, filename);
+  WriterCtx* wc = writerCtxCreate(TFile, fullname, true, 1024 * 1024);
+  if (wc == NULL) { return NULL; }
+
+  TFileReader* reader = tfileReaderCreateImpl(wc);
+  return reader;
+
+  // tfileSerialCacheKey(&key, buf);
+}
 TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header) {
   // char pathBuf[128] = {0};
   // sprintf(pathBuf, "%s/% " PRIu64 "-%d-%d.tindex", path, suid, colId, version);
@@ -249,28 +260,29 @@ TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header) {
   return tw;
 }
 
-int tfileWriterPut(TFileWriter* tw, void* data) {
+int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
   // sort by coltype and write to tindex
-  __compar_fn_t fn;
-
-  int8_t colType = tw->header.colType;
-  if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) {
-    fn = tfileStrCompare;
-  } else {
-    fn = getComparFunc(colType, 0);
+  if (order == false) {
+    __compar_fn_t fn;
+    int8_t        colType = tw->header.colType;
+    if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) {
+      fn = tfileStrCompare;
+    } else {
+      fn = getComparFunc(colType, 0);
+    }
+    taosArraySortPWithExt((SArray*)(data), tfileValueCompare, &fn);
   }
-  taosArraySortPWithExt((SArray*)(data), tfileValueCompare, &fn);
 
-  int32_t bufLimit = 4096, offset = 0;
-  char*   buf = calloc(1, sizeof(char) * bufLimit);
-  char*   p = buf;
+  int32_t bufLimit = 64 * 4096, offset = 0;
+  // char*   buf = calloc(1, sizeof(char) * bufLimit);
+  // char*   p = buf;
   int32_t sz = taosArrayGetSize((SArray*)data);
   int32_t fstOffset = tw->offset;
 
   // ugly code, refactor later
   for (size_t i = 0; i < sz; i++) {
     TFileValue* v = taosArrayGetP((SArray*)data, i);
-
+    // taosArrayRemoveDuplicate(v->tablId, tfileUidCompare, NULL);
     int32_t tbsz = taosArrayGetSize(v->tableId);
     fstOffset += TF_TABLE_TATOAL_SIZE(tbsz);
   }
@@ -282,30 +294,39 @@ int tfileWriterPut(TFileWriter* tw, void* data) {
     int32_t tbsz = taosArrayGetSize(v->tableId);
     // check buf has enough space or not
     int32_t ttsz = TF_TABLE_TATOAL_SIZE(tbsz);
-    if (offset + ttsz > bufLimit) {
-      // batch write
-      tw->ctx->write(tw->ctx, buf, offset);
-      offset = 0;
-      memset(buf, 0, bufLimit);
-      p = buf;
-    }
 
+    // if (offset + ttsz >= bufLimit) {
+    //  // batch write
+    //  indexInfo("offset: %d, ttsz: %d", offset, ttsz);
+    //  // std::cout << "offset: " << offset << std::endl;
+    //  // std::cout << "ttsz:" << ttsz < < < std::endl;
+    //  tw->ctx->write(tw->ctx, buf, offset);
+    //  offset = 0;
+    //  memset(buf, 0, bufLimit);
+    //  p = buf;
+    //}
+    // if (ttsz >= bufLimit) {
+    //}
+    char* buf = calloc(1, ttsz * sizeof(char));
+    char* p = buf;
     tfileSerialTableIdsToBuf(p, v->tableId);
-    offset += ttsz;
-    p = buf + offset;
+    tw->ctx->write(tw->ctx, buf, ttsz);
+    // offset += ttsz;
+    // p = buf + offset;
     // set up value offset
     v->offset = tw->offset;
     tw->offset += ttsz;
+    free(buf);
   }
-  if (offset != 0) {
-    // write reversed data in buf to tindex
-    tw->ctx->write(tw->ctx, buf, offset);
-  }
-  tfree(buf);
+  // if (offset != 0) {
+  // write reversed data in buf to tindex
+  // tw->ctx->write(tw->ctx, buf, offset);
+  //}
+  // tfree(buf);
 
   tw->fb = fstBuilderCreate(tw->ctx, 0);
   if (tw->fb == NULL) {
-    tfileWriterDestroy(tw);
+    tfileWriterClose(tw);
     return -1;
   }
   // write fst
@@ -321,27 +342,20 @@ int tfileWriterPut(TFileWriter* tw, void* data) {
   tw->fb = NULL;
   return 0;
 }
-void tfileWriteClose(TFileWriter* tw) {
-  if (tw == NULL) {
-    return;
-  }
-  writerCtxDestroy(tw->ctx);
+void tfileWriterClose(TFileWriter* tw) {
+  if (tw == NULL) { return; }
+  writerCtxDestroy(tw->ctx, false);
   free(tw);
 }
 void tfileWriterDestroy(TFileWriter* tw) {
-  if (tw == NULL) {
-    return;
-  }
-
-  writerCtxDestroy(tw->ctx);
+  if (tw == NULL) { return; }
+  writerCtxDestroy(tw->ctx, false);
   free(tw);
 }
 
 IndexTFile* indexTFileCreate(const char* path) {
   IndexTFile* tfile = calloc(1, sizeof(IndexTFile));
-  if (tfile == NULL) {
-    return NULL;
-  }
+  if (tfile == NULL) { return NULL; }
 
   tfile->cache = tfileCacheCreate(path);
   return tfile;
@@ -350,18 +364,14 @@ void IndexTFileDestroy(IndexTFile* tfile) { free(tfile); }
 
 int indexTFileSearch(void* tfile, SIndexTermQuery* query, SArray* result) {
   int ret = -1;
-  if (tfile == NULL) {
-    return ret;
-  }
+  if (tfile == NULL) { return ret; }
   IndexTFile* pTfile = (IndexTFile*)tfile;
 
   SIndexTerm*   term = query->term;
   TFileCacheKey key = {
       .suid = term->suid, .colType = term->colType, .colName = term->colName, .nColName = term->nColName};
   TFileReader* reader = tfileCacheGet(pTfile->cache, &key);
-  if (reader == NULL) {
-    return 0;
-  }
+  if (reader == NULL) { return 0; }
 
   return tfileReaderSearch(reader, query, result);
 }
@@ -381,9 +391,7 @@ static bool tfileIteratorNext(Iterate* iiter) {
 
   TFileFstIter*          tIter = iiter->iter;
   StreamWithStateResult* rt = streamWithStateNextWith(tIter->st, NULL);
-  if (rt == NULL) {
-    return false;
-  }
+  if (rt == NULL) { return false; }
 
   int32_t sz = 0;
   char*   ch = (char*)fstSliceData(&rt->data, &sz);
@@ -391,12 +399,9 @@ static bool tfileIteratorNext(Iterate* iiter) {
   memcpy(colVal, ch, sz);
 
   offset = (uint64_t)(rt->out.out);
-
   swsResultDestroy(rt);
   // set up iterate value
-  if (tfileReaderLoadTableIds(tIter->rdr, offset, iv->val) != 0) {
-    return false;
-  }
+  if (tfileReaderLoadTableIds(tIter->rdr, offset, iv->val) != 0) { return false; }
 
   iv->colVal = colVal;
 
@@ -407,9 +412,8 @@ static IterateValue* tifileIterateGetValue(Iterate* iter) { return &iter->val; }
 
 static TFileFstIter* tfileFstIteratorCreate(TFileReader* reader) {
   TFileFstIter* tIter = calloc(1, sizeof(Iterate));
-  if (tIter == NULL) {
-    return NULL;
-  }
+  if (tIter == NULL) { return NULL; }
+
   tIter->ctx = automCtxCreate(NULL, AUTOMATION_ALWAYS);
   tIter->fb = fstSearch(reader->fst, tIter->ctx);
   tIter->st = streamBuilderIntoStream(tIter->fb);
@@ -418,21 +422,22 @@ static TFileFstIter* tfileFstIteratorCreate(TFileReader* reader) {
 }
 
 Iterate* tfileIteratorCreate(TFileReader* reader) {
-  Iterate* iter = calloc(1, sizeof(Iterate));
+  if (reader == NULL) { return NULL; }
 
+  Iterate* iter = calloc(1, sizeof(Iterate));
   iter->iter = tfileFstIteratorCreate(reader);
   if (iter->iter == NULL) {
+    free(iter);
     return NULL;
   }
-
   iter->next = tfileIteratorNext;
   iter->getValue = tifileIterateGetValue;
+  iter->val.val = taosArrayInit(1, sizeof(uint64_t));
   return iter;
 }
 void tfileIteratorDestroy(Iterate* iter) {
-  if (iter == NULL) {
-    return;
-  }
+  if (iter == NULL) { return; }
+
   IterateValue* iv = &iter->val;
   iterateValueDestroy(iv, true);
 
@@ -445,18 +450,19 @@ void tfileIteratorDestroy(Iterate* iter) {
 }
 
 TFileReader* tfileGetReaderByCol(IndexTFile* tf, char* colName) {
-  if (tf == NULL) {
-    return NULL;
-  }
+  if (tf == NULL) { return NULL; }
   TFileCacheKey key = {.suid = 0, .colType = TSDB_DATA_TYPE_BINARY, .colName = colName, .nColName = strlen(colName)};
   return tfileCacheGet(tf->cache, &key);
 }
 
+static int tfileUidCompare(const void* a, const void* b) {
+  uint64_t l = *(uint64_t*)a;
+  uint64_t r = *(uint64_t*)b;
+  return l - r;
+}
 static int tfileStrCompare(const void* a, const void* b) {
   int ret = strcmp((char*)a, (char*)b);
-  if (ret == 0) {
-    return ret;
-  }
+  if (ret == 0) { return ret; }
   return ret < 0 ? -1 : 1;
 }
 
@@ -471,17 +477,13 @@ static int tfileValueCompare(const void* a, const void* b, const void* param) {
 
 TFileValue* tfileValueCreate(char* val) {
   TFileValue* tf = calloc(1, sizeof(TFileValue));
-  if (tf == NULL) {
-    return NULL;
-  }
-
+  if (tf == NULL) { return NULL; }
+  tf->colVal = val;
   tf->tableId = taosArrayInit(32, sizeof(uint64_t));
   return tf;
 }
 int tfileValuePush(TFileValue* tf, uint64_t val) {
-  if (tf == NULL) {
-    return -1;
-  }
+  if (tf == NULL) { return -1; }
   taosArrayPush(tf->tableId, &val);
   return 0;
 }
@@ -501,9 +503,7 @@ static void tfileSerialTableIdsToBuf(char* buf, SArray* ids) {
 static int tfileWriteFstOffset(TFileWriter* tw, int32_t offset) {
   int32_t fstOffset = offset + sizeof(tw->header.fstOffset);
   tw->header.fstOffset = fstOffset;
-  if (sizeof(fstOffset) != tw->ctx->write(tw->ctx, (char*)&fstOffset, sizeof(fstOffset))) {
-    return -1;
-  }
+  if (sizeof(fstOffset) != tw->ctx->write(tw->ctx, (char*)&fstOffset, sizeof(fstOffset))) { return -1; }
   tw->offset += sizeof(fstOffset);
   return 0;
 }
@@ -514,9 +514,7 @@ static int tfileWriteHeader(TFileWriter* writer) {
   memcpy(buf, (char*)header, sizeof(buf));
 
   int nwrite = writer->ctx->write(writer->ctx, buf, sizeof(buf));
-  if (sizeof(buf) != nwrite) {
-    return -1;
-  }
+  if (sizeof(buf) != nwrite) { return -1; }
   writer->offset = nwrite;
   return 0;
 }
@@ -547,12 +545,10 @@ static int tfileReaderLoadHeader(TFileReader* reader) {
 }
 static int tfileReaderLoadFst(TFileReader* reader) {
   // current load fst into memory, refactor it later
-  static int FST_MAX_SIZE = 16 * 1024;
+  static int FST_MAX_SIZE = 64 * 1024;
 
   char* buf = calloc(1, sizeof(char) * FST_MAX_SIZE);
-  if (buf == NULL) {
-    return -1;
-  }
+  if (buf == NULL) { return -1; }
 
   WriterCtx* ctx = reader->ctx;
   int32_t    nread = ctx->readFrom(ctx, buf, FST_MAX_SIZE, reader->header.fstOffset);
@@ -575,36 +571,33 @@ static int tfileReaderLoadTableIds(TFileReader* reader, int32_t offset, SArray*
 
   int32_t total = sizeof(uint64_t) * nid;
   char*   buf = calloc(1, total);
-  if (buf == NULL) {
-    return -1;
-  }
+  if (buf == NULL) { return -1; }
 
   nread = ctx->read(ctx, buf, total);
   assert(total == nread);
 
-  for (int32_t i = 0; i < nid; i++) {
-    taosArrayPush(result, (uint64_t*)buf + i);
-  }
+  for (int32_t i = 0; i < nid; i++) { taosArrayPush(result, (uint64_t*)buf + i); }
   free(buf);
   return 0;
 }
 void tfileReaderRef(TFileReader* reader) {
+  if (reader == NULL) { return; }
   int ref = T_REF_INC(reader);
   UNUSED(ref);
 }
 
 void tfileReaderUnRef(TFileReader* reader) {
+  if (reader == NULL) { return; }
   int ref = T_REF_DEC(reader);
   if (ref == 0) {
+    // do nothing
     tfileReaderDestroy(reader);
   }
 }
 
 static int tfileGetFileList(const char* path, SArray* result) {
   DIR* dir = opendir(path);
-  if (NULL == dir) {
-    return -1;
-  }
+  if (NULL == dir) { return -1; }
 
   struct dirent* entry;
   while ((entry = readdir(dir)) != NULL) {
@@ -632,9 +625,7 @@ static int tfileCompare(const void* a, const void* b) {
   size_t bLen = strlen(bName);
 
   int ret = strncmp(aName, bName, aLen > bLen ? aLen : bLen);
-  if (ret == 0) {
-    return ret;
-  }
+  if (ret == 0) { return ret; }
   return ret < 0 ? -1 : 1;
 }
 // tfile name suid-colId-version.tindex

source/libs/index/test/fstTest.cc

@@ -12,9 +12,7 @@
 #include "tskiplist.h"
 #include "tutil.h"
 
-void* callback(void* s) {
-  return s;
-}
+void* callback(void* s) { return s; }
 
 static std::string fileName = "/tmp/tindex.tindex";
 class FstWriter {
@@ -34,7 +32,7 @@ class FstWriter {
     fstBuilderFinish(_b);
     fstBuilderDestroy(_b);
 
-    writerCtxDestroy(_wc);
+    writerCtxDestroy(_wc, false);
   }
 
  private:
@@ -102,7 +100,7 @@ class FstReadMemory {
     fstCountingWriterDestroy(_w);
     fstDestroy(_fst);
     fstSliceDestroy(&_s);
-    writerCtxDestroy(_wc);
+    writerCtxDestroy(_wc, false);
   }
 
  private:

source/libs/index/test/indexTests.cc

@@ -57,7 +57,7 @@ class FstWriter {
     fstBuilderFinish(_b);
     fstBuilderDestroy(_b);
 
-    writerCtxDestroy(_wc);
+    writerCtxDestroy(_wc, false);
   }
 
  private:
@@ -76,9 +76,7 @@ class FstReadMemory {
   bool init() {
     char* buf = (char*)calloc(1, sizeof(char) * _size);
     int   nRead = fstCountingWriterRead(_w, (uint8_t*)buf, _size);
-    if (nRead <= 0) {
-      return false;
-    }
+    if (nRead <= 0) { return false; }
     _size = nRead;
     _s = fstSliceCreate((uint8_t*)buf, _size);
     _fst = fstCreate(&_s);
@@ -104,9 +102,7 @@ class FstReadMemory {
     StreamWithState*       st = streamBuilderIntoStream(sb);
     StreamWithStateResult* rt = NULL;
 
-    while ((rt = streamWithStateNextWith(st, NULL)) != NULL) {
-      result.push_back((uint64_t)(rt->out.out));
-    }
+    while ((rt = streamWithStateNextWith(st, NULL)) != NULL) { result.push_back((uint64_t)(rt->out.out)); }
     return true;
   }
   bool SearchWithTimeCostUs(AutomationCtx* ctx, std::vector<uint64_t>& result) {
@@ -120,7 +116,7 @@ class FstReadMemory {
     fstCountingWriterDestroy(_w);
     fstDestroy(_fst);
     fstSliceDestroy(&_s);
-    writerCtxDestroy(_wc);
+    writerCtxDestroy(_wc, true);
   }
 
  private:
@@ -182,9 +178,7 @@ void checkFstPerf() {
   delete fw;
 
   FstReadMemory* m = new FstReadMemory(1024 * 64);
-  if (m->init()) {
-    printf("success to init fst read");
-  }
+  if (m->init()) { printf("success to init fst read"); }
   Performance_fstReadRecords(m);
   delete m;
 }
@@ -346,10 +340,8 @@ class TFileObj {
       tfileReaderDestroy(reader_);
       reader_ = NULL;
     }
-    if (writer_ == NULL) {
-      InitWriter();
-    }
-    return tfileWriterPut(writer_, tv);
+    if (writer_ == NULL) { InitWriter(); }
+    return tfileWriterPut(writer_, tv, false);
   }
   bool InitWriter() {
     TFileHeader header;
@@ -388,12 +380,8 @@ class TFileObj {
     return tfileReaderSearch(reader_, query, result);
   }
   ~TFileObj() {
-    if (writer_) {
-      tfileWriterDestroy(writer_);
-    }
-    if (reader_) {
-      tfileReaderDestroy(reader_);
-    }
+    if (writer_) { tfileWriterDestroy(writer_); }
+    if (reader_) { tfileReaderDestroy(reader_); }
   }
 
  private:
@@ -465,9 +453,7 @@ TEST_F(IndexTFileEnv, test_tfile_write) {
   taosArrayPush(data, &v4);
 
   fObj->Put(data);
-  for (size_t i = 0; i < taosArrayGetSize(data); i++) {
-    destroyTFileValue(taosArrayGetP(data, i));
-  }
+  for (size_t i = 0; i < taosArrayGetSize(data); i++) { destroyTFileValue(taosArrayGetP(data, i)); }
   taosArrayDestroy(data);
 
   std::string     colName("voltage");
@@ -569,7 +555,8 @@ TEST_F(IndexCacheEnv, cache_test) {
                                        colVal.c_str(), colVal.size());
     coj->Put(term, colId, version++, suid++);
   }
-
+  coj->Debug();
+  std::cout << "--------first----------" << std::endl;
   {
     std::string colVal("v3");
     SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
@@ -582,6 +569,8 @@ TEST_F(IndexCacheEnv, cache_test) {
                                        colVal.c_str(), colVal.size());
     coj->Put(term, othColId, version++, suid++);
   }
+  coj->Debug();
+  std::cout << "--------second----------" << std::endl;
   {
     std::string colVal("v4");
     for (size_t i = 0; i < 10; i++) {
@@ -602,7 +591,7 @@ TEST_F(IndexCacheEnv, cache_test) {
     STermValueType  valType;
 
     coj->Get(&query, colId, 10000, ret, &valType);
-    // std::cout << "size : " << taosArrayGetSize(ret) << std::endl;
+    std::cout << "size : " << taosArrayGetSize(ret) << std::endl;
     assert(taosArrayGetSize(ret) == 4);
   }
   {
@@ -635,6 +624,20 @@ class IndexObj {
     }
     return ret;
   }
+  int WriteMillonData(const std::string& colName, const std::string& colVal = "Hello world",
+                      size_t numOfTable = 100 * 10000) {
+    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 < numOfTable; i++) {
+      int ret = Put(terms, i);
+      assert(ret == 0);
+    }
+    indexMultiTermDestroy(terms);
+    return numOfTable;
+  }
+
   int Put(SIndexMultiTerm* fvs, uint64_t uid) {
     numOfWrite += taosArrayGetSize(fvs);
     return indexPut(idx, fvs, uid);
@@ -645,6 +648,17 @@ class IndexObj {
     return indexSearch(idx, multiQ, result);
   }
 
+  int SearchOne(const std::string& colName, const std::string& colVal) {
+    SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
+    SIndexTerm*           term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
+                                       colVal.c_str(), colVal.size());
+    indexMultiTermQueryAdd(mq, term, QUERY_TERM);
+
+    SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t));
+    if (Search(mq, result) == 0) { std::cout << "search one successfully" << std::endl; }
+    return taosArrayGetSize(result);
+    // assert(taosArrayGetSize(result) == targetSize);
+  }
   void Debug() {
     std::cout << "numOfWrite:" << numOfWrite << std::endl;
     std::cout << "numOfRead:" << numOfRead << std::endl;
@@ -665,12 +679,13 @@ class IndexObj {
 class IndexEnv2 : public ::testing::Test {
  protected:
   virtual void SetUp() {
+    tfInit();
     index = new IndexObj();
     //
   }
   virtual void TearDown() {
-    // r
     delete index;
+    tfCleanup();
   }
   IndexObj* index;
 };
@@ -722,27 +737,39 @@ TEST_F(IndexEnv2, testIndexOpen) {
 
     SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t));
     index->Search(mq, result);
-    assert(taosArrayGetSize(result) == targetSize);
+    std::cout << "target size: " << taosArrayGetSize(result) << std::endl;
+    // assert(taosArrayGetSize(result) == targetSize);
   }
 }
-TEST_F(IndexEnv2, testIndex_CachePut) {
+
+TEST_F(IndexEnv2, testIndex_TrigeFlush) {
   std::string path = "/tmp";
-  if (index->Init(path) != 0) {
-  }
+  if (index->Init(path) != 0) {}
+  int numOfTable = 100 * 10000;
+  index->WriteMillonData("tag1", "Hello world", numOfTable);
+  int target = index->SearchOne("tag1", "Hellow world");
+  assert(numOfTable == target);
 }
-
-TEST_F(IndexEnv2, testIndexr_TFilePut) {
+TEST_F(IndexEnv2, testIndex_serarch_cache_and_tfile) {
   std::string path = "/tmp";
-  if (index->Init(path) != 0) {
-  }
+  if (index->Init(path) != 0) {}
 }
-TEST_F(IndexEnv2, testIndex_CacheSearch) {
+TEST_F(IndexEnv2, testIndex_multi_thread_write) {
   std::string path = "/tmp";
-  if (index->Init(path) != 0) {
-  }
+  if (index->Init(path) != 0) {}
 }
-TEST_F(IndexEnv2, testIndex_TFileSearch) {
+TEST_F(IndexEnv2, testIndex_multi_thread_read) {
   std::string path = "/tmp";
-  if (index->Init(path) != 0) {
-  }
+  if (index->Init(path) != 0) {}
+}
+
+TEST_F(IndexEnv2, testIndex_restart) {
+  std::string path = "/tmp";
+  if (index->Init(path) != 0) {}
+}
+
+TEST_F(IndexEnv2, testIndex_performance) {
+  std::string path = "/tmp";
+  if (index->Init(path) != 0) {}
 }
+TEST_F(IndexEnv2, testIndexMultiTag) {}