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

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

include/libs/planner/planner.h

@@ -108,7 +108,7 @@ typedef struct SProjectPhyNode {
 typedef struct SExchangePhyNode {
   SPhyNode    node;
   uint64_t    srcTemplateId; // template id of datasource suplans
-  SArray     *pSourceEpSet;  // SEpSet, scheduler fill by calling qSetSuplanExecutionNode
+  SArray     *pSrcEndPoints;  // SEpAddrMsg, scheduler fill by calling qSetSuplanExecutionNode
 } SExchangePhyNode;
 
 typedef struct SSubplanId {
@@ -129,6 +129,7 @@ typedef struct SSubplan {
 
 typedef struct SQueryDag {
   uint64_t queryId;
+  int32_t  numOfSubplans;
   SArray  *pSubplans; // Element is SArray*, and nested element is SSubplan. The execution level of subplan, starting from 0.
 } SQueryDag;
 
@@ -137,7 +138,11 @@ typedef struct SQueryDag {
  */
 int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag);
 
-int32_t qSetSubplanExecutionNode(SSubplan* subplan, SArray* eps);
+// Set datasource of this subplan, multiple calls may be made to a subplan.
+// @subplan subplan to be schedule
+// @templateId templateId of a group of datasource subplans of this @subplan
+// @eps Execution location of this group of datasource subplans, is an array of SEpAddr structures 
+int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps);
 
 int32_t qExplainQuery(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, char** str);
 

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) {
-//
-//}
+
+
 
 

source/libs/planner/inc/plannerInt.h

@@ -100,8 +100,9 @@ int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str);
 int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql);
 
 int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag);
-
+int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps);
 int32_t subPlanToString(const SSubplan *pPhyNode, char** str);
+int32_t stringToSubplan(const char* str, SSubplan** subplan);
 
 /**
  * Destroy the query plan object.

source/libs/planner/src/physicalPlan.c

@@ -19,6 +19,13 @@
 #define STORE_CURRENT_SUBPLAN(cxt) SSubplan* _ = cxt->pCurrentSubplan
 #define RECOVERY_CURRENT_SUBPLAN(cxt) cxt->pCurrentSubplan = _
 
+typedef struct SPlanContext {
+  struct SCatalog* pCatalog;
+  struct SQueryDag* pDag;
+  SSubplan* pCurrentSubplan;
+  SSubplanId nextId;
+} SPlanContext;
+
 static const char* gOpName[] = {
   "Unknown",
 #define INCLUDE_AS_NAME
@@ -26,12 +33,14 @@ static const char* gOpName[] = {
 #undef INCLUDE_AS_NAME
 };
 
-typedef struct SPlanContext {
-  struct SCatalog* pCatalog;
-  struct SQueryDag* pDag;
-  SSubplan* pCurrentSubplan;
-  SSubplanId nextId;
-} SPlanContext;
+int32_t opNameToOpType(const char* name) {
+  for (int32_t i = 1; i < sizeof(gOpName) / sizeof(gOpName[0]); ++i) {
+    if (strcmp(name, gOpName[i])) {
+      return i;
+    }
+  }
+  return OP_Unknown;
+}
 
 static void toDataBlockSchema(SQueryPlanNode* pPlanNode, SDataBlockSchema* dataBlockSchema) {
   SWAP(dataBlockSchema->pSchema, pPlanNode->pSchema, SSchema*);
@@ -216,11 +225,6 @@ int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryD
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t opNameToOpType(const char* name) {
-  for (int32_t i = 1; i < sizeof(gOpName) / sizeof(gOpName[0]); ++i) {
-    if (strcmp(name, gOpName[i])) {
-      return i;
-    }
-  }
-  return OP_Unknown;
+int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps) {
+  //todo
 }

source/libs/planner/src/physicalPlanJson.c

@@ -463,6 +463,191 @@ static bool exprInfoFromJson(const cJSON* json, void* obj) {
   return res;
 }
 
+static const char* jkTimeWindowStartKey = "StartKey";
+static const char* jkTimeWindowEndKey = "EndKey";
+
+static bool timeWindowToJson(const void* obj, cJSON* json) {
+  const STimeWindow* win = (const STimeWindow*)obj;
+  bool res = cJSON_AddNumberToObject(json, jkTimeWindowStartKey, win->skey);
+  if (res) {
+    res = cJSON_AddNumberToObject(json, jkTimeWindowEndKey, win->ekey);
+  }
+  return res;
+}
+
+static bool timeWindowFromJson(const cJSON* json, void* obj) {
+  STimeWindow* win = (STimeWindow*)obj;
+  win->skey = getNumber(json, jkTimeWindowStartKey);
+  win->ekey = getNumber(json, jkTimeWindowEndKey);
+  return true;
+}
+
+static const char* jkScanNodeTableId = "TableId";
+static const char* jkScanNodeTableType = "TableType";
+
+static bool scanNodeToJson(const void* obj, cJSON* json) {
+  const SScanPhyNode* scan = (const SScanPhyNode*)obj;
+  bool res = cJSON_AddNumberToObject(json, jkScanNodeTableId, scan->uid);
+  if (res) {
+    res = cJSON_AddNumberToObject(json, jkScanNodeTableType, scan->tableType);
+  }
+  return res;
+}
+
+static bool scanNodeFromJson(const cJSON* json, void* obj) {
+  SScanPhyNode* scan = (SScanPhyNode*)obj;
+  scan->uid = getNumber(json, jkScanNodeTableId);
+  scan->tableType = getNumber(json, jkScanNodeTableType);
+  return true;
+}
+
+static const char* jkTableScanNodeFlag = "Flag";
+static const char* jkTableScanNodeWindow = "Window";
+static const char* jkTableScanNodeTagsConditions = "TagsConditions";
+
+static bool tableScanNodeToJson(const void* obj, cJSON* json) {
+  const STableScanPhyNode* scan = (const STableScanPhyNode*)obj;
+  bool res = scanNodeToJson(obj, json);
+  if (res) {
+    res = cJSON_AddNumberToObject(json, jkTableScanNodeFlag, scan->scanFlag);
+  }
+  if (res) {
+    res = addObject(json, jkTableScanNodeWindow, timeWindowToJson, &scan->window);
+  }
+  if (res) {
+    res = addArray(json, jkTableScanNodeTagsConditions, exprInfoToJson, scan->pTagsConditions);
+  }
+  return res;
+}
+
+static bool tableScanNodeFromJson(const cJSON* json, void* obj) {
+  STableScanPhyNode* scan = (STableScanPhyNode*)obj;
+  bool res = scanNodeFromJson(json, obj);
+  if (res) {
+    scan->scanFlag = getNumber(json, jkTableScanNodeFlag);
+  }
+  if (res) {
+    res = fromObject(json, jkTableScanNodeWindow, timeWindowFromJson, &scan->window, true);
+  }
+  if (res) {
+    res = fromArray(json, jkTableScanNodeTagsConditions, exprInfoFromJson, &scan->pTagsConditions, sizeof(SExprInfo));
+  }
+  return res;
+}
+
+static const char* jkEpAddrFqdn = "Fqdn";
+static const char* jkEpAddrPort = "Port";
+
+static bool epAddrToJson(const void* obj, cJSON* json) {
+  const SEpAddrMsg* ep = (const SEpAddrMsg*)obj;
+  bool res = cJSON_AddStringToObject(json, jkEpAddrFqdn, ep->fqdn);
+  if (res) {
+    res = cJSON_AddNumberToObject(json, jkEpAddrPort, ep->port);
+  }
+  return res;
+}
+
+static bool epAddrFromJson(const cJSON* json, void* obj) {
+  SEpAddrMsg* ep = (SEpAddrMsg*)obj;
+  copyString(json, jkEpAddrFqdn, ep->fqdn);
+  ep->port = getNumber(json, jkEpAddrPort);
+  return true;
+}
+
+static const char* jkExchangeNodeSrcTemplateId = "SrcTemplateId";
+static const char* jkExchangeNodeSrcEndPoints = "SrcEndPoints";
+
+static bool exchangeNodeToJson(const void* obj, cJSON* json) {
+  const SExchangePhyNode* exchange = (const SExchangePhyNode*)obj;
+  bool res = cJSON_AddNumberToObject(json, jkExchangeNodeSrcTemplateId, exchange->srcTemplateId);
+  if (res) {
+    res = addArray(json, jkExchangeNodeSrcEndPoints, epAddrToJson, exchange->pSrcEndPoints);
+  }
+  return res;
+}
+
+static bool exchangeNodeFromJson(const cJSON* json, void* obj) {
+  SExchangePhyNode* exchange = (SExchangePhyNode*)obj;
+  exchange->srcTemplateId = getNumber(json, jkExchangeNodeSrcTemplateId);
+  return fromArray(json, jkExchangeNodeSrcEndPoints, epAddrFromJson, &exchange->pSrcEndPoints, sizeof(SEpAddrMsg));
+}
+
+static bool specificPhyNodeToJson(const void* obj, cJSON* json) {
+  const SPhyNode* phyNode = (const SPhyNode*)obj;
+  switch (phyNode->info.type) {
+    case OP_TableScan:
+    case OP_DataBlocksOptScan:
+    case OP_TableSeqScan:
+      return tableScanNodeToJson(obj, json);
+    case OP_TagScan:
+    case OP_SystemTableScan:
+      return scanNodeToJson(obj, json);
+    case OP_Aggregate:
+      break; // todo
+    case OP_Project:
+      return true;
+    case OP_Groupby:
+    case OP_Limit:
+    case OP_SLimit:
+    case OP_TimeWindow:
+    case OP_SessionWindow:
+    case OP_StateWindow:
+    case OP_Fill:
+    case OP_MultiTableAggregate:
+    case OP_MultiTableTimeInterval:
+    case OP_Filter:
+    case OP_Distinct:
+    case OP_Join:
+    case OP_AllTimeWindow:
+    case OP_AllMultiTableTimeInterval:
+    case OP_Order:
+      break; // todo
+    case OP_Exchange:
+      return exchangeNodeToJson(obj, json);
+    default:
+      break;
+  }
+  return false;
+}
+
+static bool specificPhyNodeFromJson(const cJSON* json, void* obj) {
+  SPhyNode* phyNode = (SPhyNode*)obj;
+  switch (phyNode->info.type) {
+    case OP_TableScan:
+    case OP_DataBlocksOptScan:
+    case OP_TableSeqScan:
+      return tableScanNodeFromJson(json, obj);
+    case OP_TagScan:
+    case OP_SystemTableScan:
+      return scanNodeFromJson(json, obj);
+    case OP_Aggregate:
+      break; // todo
+    case OP_Project:
+      return true;
+    case OP_Groupby:
+    case OP_Limit:
+    case OP_SLimit:
+    case OP_TimeWindow:
+    case OP_SessionWindow:
+    case OP_StateWindow:
+    case OP_Fill:
+    case OP_MultiTableAggregate:
+    case OP_MultiTableTimeInterval:
+    case OP_Filter:
+    case OP_Distinct:
+    case OP_Join:
+    case OP_AllTimeWindow:
+    case OP_AllMultiTableTimeInterval:
+    case OP_Order:
+      break; // todo
+    case OP_Exchange:
+      return exchangeNodeFromJson(json, obj);
+    default:
+      break;
+  }
+  return false;
+}
+
 static const char* jkPnodeName = "Name";
 static const char* jkPnodeTargets = "Targets";
 static const char* jkPnodeConditions = "Conditions";
@@ -484,6 +669,9 @@ static bool phyNodeToJson(const void* obj, cJSON* jNode) {
   if (res) {
     res = addArray(jNode, jkPnodeChildren, phyNodeToJson, phyNode->pChildren);
   }
+  if (res) {
+    res = addObject(jNode, phyNode->info.name, specificPhyNodeToJson, phyNode);
+  }
   return res;
 }
 
@@ -501,6 +689,9 @@ static bool phyNodeFromJson(const cJSON* json, void* obj) {
   if (res) {
     res = fromArray(json, jkPnodeChildren, phyNodeFromJson, &node->pChildren, sizeof(SSlotSchema));
   }
+  if (res) {
+    res = fromObject(json, node->info.name, specificPhyNodeFromJson, node, true);
+  }
   return res;
 }
 

source/libs/planner/src/planner.c

@@ -46,6 +46,10 @@ int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet*
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps) {
+  return setSubplanExecutionNode(subplan, templateId, eps);
+}
+
 int32_t qSubPlanToString(const SSubplan *subplan, char** str) {
   return subPlanToString(subplan, str);
 }