Merge pull request #9043 from taosdata/feature/fst

update fst search frame

source/libs/index/inc/index_fst.h

@@ -291,11 +291,11 @@ typedef struct StreamState {
 void streamStateDestroy(void *s);
 
 typedef struct StreamWithState {
-  Fst        *fst;
-  Automation *aut; 
-  SArray     *inp;
-  FstOutput   emptyOutput;
-  SArray     *stack; // <StreamState>
+  Fst           *fst;
+  AutomationCtx *aut; 
+  SArray        *inp;
+  FstOutput     emptyOutput;
+  SArray        *stack; // <StreamState>
   FstBoundWithData *endAt;
 } StreamWithState ;
 
@@ -310,19 +310,19 @@ StreamWithStateResult *swsResultCreate(FstSlice *data, FstOutput fOut, void *sta
 void swsResultDestroy(StreamWithStateResult *result);
 
 typedef void* (*StreamCallback)(void *);
-StreamWithState *streamWithStateCreate(Fst *fst, Automation *automation, FstBoundWithData *min, FstBoundWithData *max) ;
+StreamWithState *streamWithStateCreate(Fst *fst, AutomationCtx *automation, FstBoundWithData *min, FstBoundWithData *max) ;
 void streamWithStateDestroy(StreamWithState *sws);
 bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min);           
 StreamWithStateResult* streamWithStateNextWith(StreamWithState *sws, StreamCallback callback);
 
 typedef struct FstStreamBuilder {
   Fst *fst; 
-  Automation *aut;
+  AutomationCtx *aut;
   FstBoundWithData *min;  
   FstBoundWithData *max;
 } FstStreamBuilder;
 
-FstStreamBuilder *fstStreamBuilderCreate(Fst *fst, Automation *aut); 
+FstStreamBuilder *fstStreamBuilderCreate(Fst *fst, AutomationCtx *aut); 
 // set up bound range
 // refator, simple code by marco 
 

source/libs/index/inc/index_fst_automation.h

@@ -19,33 +19,39 @@
 extern "C" {
 #endif
 
+#include "index_fst_util.h"
 typedef struct AutomationCtx AutomationCtx;
 
+typedef enum AutomationType {
+  AUTOMATION_PREFIX,
+  AUTMMATION_MATCH 
+} AutomationType;
+
 typedef struct StartWith {
   AutomationCtx  *autoSelf;
 } StartWith;
 
 typedef struct Complement {
   AutomationCtx *autoSelf;
-  
 } Complement;
 
 // automation 
 typedef struct AutomationCtx {
-// automation interface
-  void *data;
+  AutomationType type; 
 } AutomationCtx;
 
-typedef struct Automation {
-  void* (*start)() ; 
-  bool (*isMatch)(void *);
-  bool (*canMatch)(void *data);
-  bool (*willAlwaysMatch)(void *state); 
-  void* (*accept)(void *state, uint8_t byte);
-  void* (*acceptEof)(void *state);
-  void *data;
-} Automation; 
+typedef struct AutomationFunc {
+  void* (*start)(AutomationCtx *ctx) ; 
+  bool (*isMatch)(AutomationCtx *ctx, void *);
+  bool (*canMatch)(AutomationCtx *ctx, void *data);
+  bool (*willAlwaysMatch)(AutomationCtx *ctx, void *state); 
+  void* (*accept)(AutomationCtx *ctx, void *state, uint8_t byte);
+  void* (*acceptEof)(AutomationCtx *ct, void *state);
+} AutomationFunc; 
+
+AutomationCtx *automCtxCreate(void *data, AutomationType type);
 
+extern AutomationFunc automFuncs[]; 
 #ifdef __cplusplus
 }
 #endif

source/libs/index/src/index_fst.c

@@ -1177,7 +1177,7 @@ void fstBoundDestroy(FstBoundWithData *bound) {
   free(bound);
 }
 
-StreamWithState *streamWithStateCreate(Fst *fst, Automation *automation, FstBoundWithData *min, FstBoundWithData *max) {
+StreamWithState *streamWithStateCreate(Fst *fst, AutomationCtx *automation, FstBoundWithData *min, FstBoundWithData *max) {
   StreamWithState *sws = calloc(1, sizeof(StreamWithState));
   if (sws == NULL) { return NULL; } 
 
@@ -1204,6 +1204,8 @@ void streamWithStateDestroy(StreamWithState *sws) {
 }
 
 bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
+
+  AutomationCtx *aut = sws->aut;
   if (fstBoundWithDataIsEmpty(min)) {
     if (fstBoundWithDataIsIncluded(min)) {
        sws->emptyOutput.out = fstEmptyFinalOutput(sws->fst, &(sws->emptyOutput.null));
@@ -1211,7 +1213,7 @@ bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
     StreamState s = {.node     = fstGetRoot(sws->fst),
                      .trans    = 0,
                      .out      = {.null = false, .out = 0},
-                     .autState = sws->aut->start()}; // auto.start callback 
+                     .autState = automFuncs[aut->type].start(aut)}; // auto.start callback 
     taosArrayPush(sws->stack, &s);
     return true;
   } 
@@ -1229,7 +1231,8 @@ bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
 
   FstNode *node = fstGetRoot(sws->fst); 
   Output  out = 0;
-  void*   autState = sws->aut->start();  
+  //void*   autState = sws->aut->start();  
+  void* autState = automFuncs[aut->type].start(aut);
 
   int32_t len; 
   uint8_t *data = fstSliceData(key, &len);  
@@ -1241,7 +1244,8 @@ bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
       FstTransition trn;
       fstNodeGetTransitionAt(node, res, &trn);   
       void *preState = autState;
-      autState = sws->aut->accept(preState, b);
+      // autState = sws->aut->accept(preState, b);
+      autState = automFuncs[aut->type].accept(aut, preState, b);
       taosArrayPush(sws->inp, &b);
       StreamState s = {.node     = node, 
                        .trans    = res + 1, 
@@ -1298,6 +1302,7 @@ bool streamWithStateSeekMin(StreamWithState *sws, FstBoundWithData *min) {
 }          
 
 StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallback callback) {
+  AutomationCtx *aut = sws->aut;
   FstOutput output = sws->emptyOutput; 
   if (output.null == false) {
     FstSlice emptySlice = fstSliceCreate(NULL, 0);   
@@ -1306,15 +1311,15 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
       sws->stack = (SArray *)taosArrayInit(256, sizeof(StreamState)); 
       return NULL;
     }
-    void* start = sws->aut->start();
-    if (sws->aut->isMatch(start)) { 
+    void *start = automFuncs[aut->type].start(aut);
+    if (automFuncs[aut->type].isMatch(aut, start)) {
       FstSlice s = fstSliceCreate(NULL, 0);
       return swsResultCreate(&s, output, callback(start));
     }
   }
   while (taosArrayGetSize(sws->stack) > 0) {
     StreamState *p = (StreamState *)taosArrayPop(sws->stack);     
-    if (p->trans >= FST_NODE_LEN(p->node) || !sws->aut->canMatch(p->autState)) {
+    if (p->trans >= FST_NODE_LEN(p->node) || automFuncs[aut->type].canMatch(aut, p->autState)) {
       if (FST_NODE_ADDR(p->node) != fstGetRootAddr(sws->fst)) {
         taosArrayPop(sws->inp);
       }
@@ -1324,16 +1329,18 @@ StreamWithStateResult *streamWithStateNextWith(StreamWithState *sws, StreamCallb
     FstTransition trn; 
     fstNodeGetTransitionAt(p->node, p->trans, &trn);
     Output out = p->out.out + trn.out;
-    void* nextState = sws->aut->accept(p->autState, trn.inp);
+    void* nextState = automFuncs[aut->type].accept(aut, p->autState, trn.inp);
     void* tState = callback(nextState);
-    bool isMatch = sws->aut->isMatch(nextState);
+    bool isMatch = automFuncs[aut->type].isMatch(aut, nextState);
+    //bool isMatch = sws->aut->isMatch(nextState);
     FstNode *nextNode = fstGetNode(sws->fst, trn.addr); 
     taosArrayPush(sws->inp, &(trn.inp)); 
 
     if (FST_NODE_IS_FINAL(nextNode)) {
-      void *eofState = sws->aut->acceptEof(nextState); 
+      //void *eofState = sws->aut->acceptEof(nextState); 
+      void *eofState = automFuncs[aut->type].acceptEof(aut, nextState);
       if (eofState != NULL) {
-        isMatch = sws->aut->isMatch(eofState); 
+        isMatch = automFuncs[aut->type].isMatch(aut, eofState);
       }
     } 
     StreamState s1 = { .node = p->node, .trans = p->trans + 1, .out = p->out, .autState = p->autState};  
@@ -1391,7 +1398,7 @@ void streamStateDestroy(void *s) {
   //free(s->autoState);
 }
 
-FstStreamBuilder *fstStreamBuilderCreate(Fst *fst, Automation *aut) {
+FstStreamBuilder *fstStreamBuilderCreate(Fst *fst, AutomationCtx *aut) {
   FstStreamBuilder *b = calloc(1, sizeof(FstStreamBuilder));
   if (NULL == b) { return NULL; }
 

source/libs/index/src/index_fst_automation.c

@@ -13,3 +13,77 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "index_fst_automation.h"
+
+
+// prefix query, impl later
+static void* prefixStart(AutomationCtx *ctx) {
+  return NULL;
+};
+static bool prefixIsMatch(AutomationCtx *ctx, void *data) {
+  return true;
+} 
+static bool prefixCanMatch(AutomationCtx *ctx, void *data) {
+  return true;
+}
+static bool prefixWillAlwaysMatch(AutomationCtx *ctx, void *state) {
+  return true;
+}
+static void* prefixAccept(AutomationCtx *ctx, void *state, uint8_t byte) {
+  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* patternAccept(AutomationCtx *ctx, void *state, uint8_t byte) {
+  return NULL;
+}
+
+static void* patternAcceptEof(AutomationCtx *ctx, void *state) {
+  return NULL;
+}
+
+AutomationFunc automFuncs[]  = {{
+    prefixStart,        
+    prefixIsMatch, 
+    prefixCanMatch,
+    prefixWillAlwaysMatch,
+    prefixAccept,
+    prefixAcceptEof
+  },  
+  {
+    patternStart,
+    patternIsMatch,
+    patternCanMatch,
+    patternWillAlwaysMatch,
+    patternAccept,
+    patternAcceptEof
+  }
+};
+
+AutomationCtx* automCtxCreate(void *data, AutomationType type) {
+  AutomationCtx *ctx =  calloc(1, sizeof(AutomationCtx));
+  if (ctx == NULL) { return NULL; }
+
+  ctx->type = type;
+  if (type == AUTOMATION_PREFIX) {
+
+  }
+} 
+

source/libs/index/test/indexTests.cpp

@@ -65,6 +65,7 @@ class FstReadMemory {
     
    ~FstReadMemory() {
     fstCountingWriterDestroy(_w);
+    fstDestroy(_fst);
     fstSliceDestroy(&_s);
   } 
   
@@ -129,10 +130,12 @@ class FstReadMemory {
 //}
 
 
+#define L 100
+#define M 100
+#define N 100
 
 int Performance_fstWriteRecords(FstWriter *b) {
   std::string str("aa"); 
-  int L = 100, M = 100, N = 10;
   for (int i = 0; i < L; i++) {
     str[0] = 'a' + i;
     str.resize(2); 
@@ -150,22 +153,29 @@ int Performance_fstWriteRecords(FstWriter *b) {
 }
 
 void Performance_fstReadRecords(FstReadMemory *m) {
-   std::string str("a");
-   for (int i = 0; i < 50; i++) {
-     //std::string str("aa"); 
-     str.push_back('a');
-     uint64_t out, cost;
-     bool ok = m->GetWithTimeCostUs(str, &out, &cost); 
-     if (ok == true) {
-      printf("success to get (%s, %" PRId64"), time cost: %" PRId64")\n", str.c_str(), out, cost);
-     } else {
-      printf("failed to get(%s)\n", str.c_str());
-     }
-   }
+  std::string str("aa");
+  for (int i = 0; i < M; i++) {
+    str[0] = 'a' + i;
+    str.resize(2); 
+    for(int j = 0; j < N; j++) {
+      str[1] = 'a' + j;
+      str.resize(2);
+      for (int k = 0; k < L; k++) {
+        str.push_back('a');
+        uint64_t val, cost; 
+        if (m->GetWithTimeCostUs(str, &val, &cost)) {
+          printf("succes to get kv(%s, %" PRId64"), cost: %" PRId64"\n", str.c_str(), val, cost);
+        } else {
+          printf("failed to get key: %s\n", str.c_str());
+        }
+      }
+    } 
+  }
 }
 void checkFstPerf() {
   FstWriter *fw = new FstWriter;
   int64_t s = taosGetTimestampUs();
+
   int num = Performance_fstWriteRecords(fw);
   int64_t e = taosGetTimestampUs();
   printf("write %d record cost %" PRId64"us\n", num,  e - s);
@@ -173,13 +183,11 @@ void checkFstPerf() {
 
   FstReadMemory *m = new FstReadMemory(1024 * 64);
   if (m->init()) {
-    uint64_t val;
-    if(m->Get("aaaaaaa", &val)) {
-      std::cout << "succes to Get val: " << val << std::endl;
-    } else {
-      std::cout << "failed to Get " << std::endl;
-    } 
+    printf("success to init fst read");  
   }  
+  Performance_fstReadRecords(m); 
+   
+  delete m;
 }