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

docs/user_manual/Small_Materialized_Aggrates.md

+# Small Materialized Aggragates
+
+**SMA** (**S**mall **M**aterialized **A**ggrates) is used to speed up the query process on materialized data cube in TDengine. TDengine 3.0 gives more flexibility on the SMA configurations.
+
+There are two kinds of SMA in TDengine:
+1. Block-wise SMA
+2. Time-range-wise SMA
+
+<!-- 
+```plantuml
+    @startmindmap mind_map_test
+    * SMA
+    ** Block-wise SMA
+    ** Time-range-wise SMA
+    @endmindmap
+```  -->
+
+![SMA in TDengine 3.0](http://www.plantuml.com/plantuml/png/Kr1GK70eBaaiAidDp4l9JInG0D7nG4PyIMfn2HTGMa5B8TZN4SBIKd3AoK_ErYtFB4v55Wt9p4tLBKhCIqz5bN981HeACHW0)
+## Block-wise SMA
+Block-wise SMA is created by default when the data are committed. Since time-series data are saved as block data in files, a corresponding SMA is create when the data block is written. The default block-wise SMA includes:
+1. sum(*)
+2. max(*)
+3. min(*)
+
+By default, the system will create SMA for each column except those columns with type *binary* and *nchar*. However, users can change the behavior by the keyword **NOSMA** to disable the SMA for a certain column like below:
+```SQL
+# create a super table with the SMA on column b disabled
+create table st (ts timestamp, a int, b int NOSMA, c double) tags (tg1 binary(10), tg2 int);
+```
+
+## Time-range-wise SMA
+In addition to the default block-wise SMA, users can create their own SMAs ondemand. Below is an example to create a SMA.
+```SQL
+# create a SMA every 10 minutes with SMA of sum, max and min
+create sma_indx sma_10min on st (sum(*), max(*), min(*), twa(*)) interval(10m);
+```
+Users can also drop a time-range-wise SMA like below:
+```SQL
+# drop the sma index
+drop sma_index sma_5min on st;
+```
+**NOTE: Creating an SMA index is a heavy operation which may take a long time and block the write operation. So create the time-range-wise SMA when creating the table or when there are not too much data.**
\ No newline at end of file

include/common/tvariant.h

@@ -27,8 +27,8 @@ typedef struct SVariant {
   uint32_t   nType;
   int32_t    nLen;  // only used for string, for number, it is useless
   union {
-    int64_t  i64;
-    uint64_t u64;
+    int64_t  i;
+    uint64_t u;
     double   d;
     char    *pz;
     wchar_t *wpz;

include/libs/function/function.h

@@ -186,7 +186,6 @@ typedef struct SResultDataInfo {
   int32_t intermediateBytes;
 } SResultDataInfo;
 
-
 typedef struct SMultiFunctionsDesc {
   bool stableQuery;
   bool groupbyColumn;
@@ -203,6 +202,8 @@ typedef struct SMultiFunctionsDesc {
   bool timewindow;
   bool topbotQuery;
   bool interpQuery;
+  bool distinct;
+  bool join;
 } SMultiFunctionsDesc;
 
 int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, SResultDataInfo* pInfo, int16_t extLength,
@@ -222,6 +223,8 @@ const char* qGetFunctionName(int32_t functionId);
 
 void extractFunctionDesc(SArray* pFunctionIdList, SMultiFunctionsDesc* pDesc);
 
+tExprNode* exprdup(tExprNode* pTree);
+
 #ifdef __cplusplus
 }
 #endif

include/libs/parser/parser.h

@@ -163,7 +163,6 @@ typedef struct SQueryStmtInfo {
   int64_t          vgroupLimit;    // table limit in case of super table projection query + global order + limit
 
   int32_t          udColumnId;    // current user-defined constant output field column id, monotonically decreases from TSDB_UD_COLUMN_INDEX
-  bool             distinct;   // distinct tag or not
   int32_t          bufLen;
   char*            buf;
   SArray          *pUdfInfo;
@@ -231,6 +230,8 @@ SExprInfo* createExprInfo(STableMetaInfo* pTableMetaInfo, int16_t functionId, SC
 int32_t copyExprInfoList(SArray* dst, const SArray* src, uint64_t uid, bool deepcopy);
 
 STableMetaInfo* getMetaInfo(SQueryStmtInfo* pQueryInfo, int32_t tableIndex);
+SSchema *getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex);
+
 int32_t getNewResColId();
 
 #ifdef __cplusplus

include/libs/planner/planner.h

@@ -56,7 +56,7 @@ int32_t qOptimizeQueryPlan(struct SQueryPlanNode* pQueryNode);
  * @param pQueryNode
  * @return
  */
-int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode* pQueryNode);
+int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode);
 
 /**
  * Convert the query plan to string, in order to display it in the shell.

include/server/vnode/tq/tq.h

@@ -17,113 +17,114 @@
 #define _TD_TQ_H_
 
 #include "os.h"
+#include "tutil.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-typedef struct tmqMsgHead {
+typedef struct TmqMsgHead {
   int32_t protoVer;
   int32_t msgType;
   int64_t cgId;
   int64_t clientId;
-} tmqMsgHead;
+} TmqMsgHead;
 
-typedef struct tmqOneAck {
+typedef struct TmqOneAck {
   int64_t topicId;
   int64_t consumeOffset;
-} tmqOneAck;
+} TmqOneAck;
 
-typedef struct tmqAcks {
+typedef struct TmqAcks {
   int32_t ackNum;
   //should be sorted
-  tmqOneAck acks[];
-} tmqAcks;
+  TmqOneAck acks[];
+} TmqAcks;
 
 //TODO: put msgs into common
-typedef struct tmqConnectReq {
-  tmqMsgHead head;
-  tmqAcks acks;
-} tmqConnectReq;
+typedef struct TmqConnectReq {
+  TmqMsgHead head;
+  TmqAcks acks;
+} TmqConnectReq;
 
-typedef struct tmqConnectRsp {
-  tmqMsgHead head;
+typedef struct TmqConnectRsp {
+  TmqMsgHead head;
   int8_t status;
-} tmqConnectRsp;
+} TmqConnectRsp;
 
-typedef struct tmqDisconnectReq {
-  tmqMsgHead head;
-} tmqDisconnectReq;
+typedef struct TmqDisconnectReq {
+  TmqMsgHead head;
+} TmqDiscconectReq;
 
-typedef struct tmqDisconnectRsp {
-  tmqMsgHead head;
+typedef struct TmqDisconnectRsp {
+  TmqMsgHead head;
   int8_t status;
-} tmqDiconnectRsp;
+} TmqDisconnectRsp;
 
-typedef struct tmqConsumeReq {
-  tmqMsgHead head;
-  tmqAcks acks;
-} tmqConsumeReq;
+typedef struct TmqConsumeReq {
+  TmqMsgHead head;
+  TmqAcks acks;
+} TmqConsumeReq;
 
-typedef struct tmqMsgContent {
+typedef struct TmqMsgContent {
   int64_t topicId;
   int64_t msgLen;
   char    msg[];
-} tmqMsgContent;
+} TmqMsgContent;
 
-typedef struct tmqConsumeRsp {
-  tmqMsgHead    head;
+typedef struct TmqConsumeRsp {
+  TmqMsgHead    head;
   int64_t       bodySize;
-  tmqMsgContent msgs[];
-} tmqConsumeRsp;
+  TmqMsgContent msgs[];
+} TmqConsumeRsp;
 
-typedef struct tmqMnodeSubscribeReq {
-  tmqMsgHead head;
-  int64_t topicLen;
-  char topic[];
-} tmqSubscribeReq;
+typedef struct TmqSubscribeReq {
+  TmqMsgHead head;
+  int32_t topicNum;
+  int64_t topic[];
+} TmqSubscribeReq;
 
-typedef struct tmqMnodeSubscribeRsp {
-  tmqMsgHead head;
+typedef struct tmqSubscribeRsp {
+  TmqMsgHead head;
   int64_t vgId;
-  char ep[]; //TSDB_EP_LEN
-} tmqSubscribeRsp;
+  char ep[TSDB_EP_LEN]; //TSDB_EP_LEN
+} TmqSubscribeRsp;
 
-typedef struct tmqHeartbeatReq {
+typedef struct TmqHeartbeatReq {
 
-} tmqHeartbeatReq;
+} TmqHeartbeatReq;
 
-typedef struct tmqHeartbeatRsp {
+typedef struct TmqHeartbeatRsp {
 
-} tmqHeartbeatRsp;
+} TmqHeartbeatRsp;
 
-typedef struct tqTopicVhandle {
-  //name
-  //
+typedef struct TqTopicVhandle {
+  int64_t topicId;
   //executor for filter
-  //
+  void*  filterExec;
   //callback for mnode
-  //
-} tqTopicVhandle;
+  //trigger when vnode list associated topic change
+  void* (*mCallback)(void*, void*);
+} TqTopicVhandle;
 
 typedef struct STQ {
   //the collection of group handle
-
+  //the handle of kvstore
 } STQ;
 
 #define TQ_BUFFER_SIZE 8
 
 //TODO: define a serializer and deserializer
-typedef struct tqBufferItem {
+typedef struct TqBufferItem {
   int64_t offset;
   //executors are identical but not concurrent
   //so it must be a copy in each item
   void* executor;
   int64_t size;
   void* content;
-} tqBufferItem;
+} TqBufferItem;
 
-typedef struct tqBufferHandle {
+typedef struct TqBufferHandle {
   //char* topic; //c style, end with '\0'
   //int64_t cgId;
   //void* ahandle;
@@ -131,32 +132,32 @@ typedef struct tqBufferHandle {
   int64_t topicId;
   int32_t head;
   int32_t tail;
-  tqBufferItem buffer[TQ_BUFFER_SIZE];
-} tqBufferHandle;
+  TqBufferItem buffer[TQ_BUFFER_SIZE];
+} TqBufferHandle;
 
-typedef struct tqListHandle {
-  tqBufferHandle* bufHandle;
-  struct tqListHandle* next;
-} tqListHandle;
+typedef struct TqListHandle {
+  TqBufferHandle bufHandle;
+  struct TqListHandle* next;
+} TqListHandle;
 
-typedef struct tqGroupHandle {
+typedef struct TqGroupHandle {
   int64_t cId;
   int64_t cgId;
   void* ahandle;
   int32_t topicNum;
-  tqListHandle *head; 
-} tqGroupHandle;
+  TqListHandle *head; 
+} TqGroupHandle;
 
-typedef struct tqQueryExec {
+typedef struct TqQueryExec {
   void* src;
-  tqBufferItem* dest;
+  TqBufferItem* dest;
   void* executor;
-} tqQueryExec;
+} TqQueryExec;
 
-typedef struct tqQueryMsg {
-  tqQueryExec *exec;
-  struct tqQueryMsg *next;
-} tqQueryMsg;
+typedef struct TqQueryMsg {
+  TqQueryExec *exec;
+  struct TqQueryMsg *next;
+} TqQueryMsg;
 
 //init in each vnode
 STQ* tqInit(void* ref_func(void*), void* unref_func(void*));
@@ -166,32 +167,30 @@ void tqCleanUp(STQ*);
 int tqPushMsg(STQ*, void* msg, int64_t version);
 int tqCommit(STQ*);
 
-int tqConsume(STQ*, tmqConsumeReq*);
+int tqConsume(STQ*, TmqConsumeReq*);
 
-tqGroupHandle* tqGetGroupHandle(STQ*, int64_t cId);
+TqGroupHandle* tqGetGroupHandle(STQ*, int64_t cId);
 
 int tqOpenTCGroup(STQ*, int64_t topicId, int64_t cgId, int64_t cId);
 int tqCloseTCGroup(STQ*, int64_t topicId, int64_t cgId, int64_t cId);
-int tqMoveOffsetToNext(tqGroupHandle*);
+int tqMoveOffsetToNext(TqGroupHandle*);
 int tqResetOffset(STQ*, int64_t topicId, int64_t cgId, int64_t offset);
-int tqRegisterContext(tqGroupHandle*, void*);
-int tqLaunchQuery(tqGroupHandle*);
-int tqSendLaunchQuery(STQ*, int64_t topicId, int64_t cgId, void* query);
-
-int tqSerializeGroupHandle(tqGroupHandle *gHandle, void** ppBytes, int32_t offset);
-int tqSerializeListHandle(tqListHandle *listHandle, void** ppBytes, int32_t offset);
-int tqSerializeBufHandle(tqBufferHandle *bufHandle, void** ppBytes, int32_t offset);
-int tqSerializeBufItem(tqBufferItem *bufItem, void** ppBytes, int32_t offset);
-
-int tqDeserializeGroupHandle(const void* pBytes, tqGroupHandle **pGhandle);
-int tqDeserializeListHandle(const void* pBytes, tqListHandle **pListHandle);
-int tqDeserializeBufHandle(const void* pBytes, tqBufferHandle **pBufHandle);
-int tqDeserializeBufItem(const void* pBytes, tqBufferItem **pBufItem);
-
-int tqGetGHandleSSize(const tqGroupHandle *gHandle);
-int tqListHandleSSize(const tqListHandle *listHandle);
-int tqBufHandleSSize(const tqBufferHandle *bufHandle);
-int tqBufItemSSize(const tqBufferItem *bufItem);
+int tqRegisterContext(TqGroupHandle*, void* ahandle);
+int tqLaunchQuery(TqGroupHandle*);
+int tqSendLaunchQuery(TqGroupHandle*);
+
+int tqSerializeGroupHandle(TqGroupHandle *gHandle, void** ppBytes);
+void* tqSerializeListHandle(TqListHandle *listHandle, void* ptr);
+void* tqSerializeBufHandle(TqBufferHandle *bufHandle, void* ptr);
+void* tqSerializeBufItem(TqBufferItem *bufItem, void* ptr);
+
+const void* tqDeserializeGroupHandle(const void* pBytes, TqGroupHandle *ghandle);
+const void* tqDeserializeBufHandle(const void* pBytes, TqBufferHandle *bufHandle);
+const void* tqDeserializeBufItem(const void* pBytes, TqBufferItem *bufItem);
+
+int tqGetGHandleSSize(const TqGroupHandle *gHandle);
+int tqBufHandleSSize();
+int tqBufItemSSize();
 
 #ifdef __cplusplus
 }

source/common/src/tvariant.c

@@ -82,9 +82,9 @@ void taosVariantCreate(SVariant *pVar, char* z, int32_t n, int32_t type) {
   switch (type) {
     case TSDB_DATA_TYPE_BOOL: {
       if (strncasecmp(z, "true", 4) == 0) {
-        pVar->i64 = TSDB_TRUE;
+        pVar->i = TSDB_TRUE;
       } else if (strncasecmp(z, "false", 5) == 0) {
-        pVar->i64 = TSDB_FALSE;
+        pVar->i = TSDB_FALSE;
       } else {
         return;
       }
@@ -106,7 +106,7 @@ void taosVariantCreate(SVariant *pVar, char* z, int32_t n, int32_t type) {
         base = 2;
       }
 
-      ret = toInteger(z, n, base, &pVar->i64, &sign);
+      ret = toInteger(z, n, base, &pVar->i, &sign);
       if (ret != 0) {
         pVar->nType = -1;   // -1 means error type
         return;
@@ -127,7 +127,7 @@ void taosVariantCreate(SVariant *pVar, char* z, int32_t n, int32_t type) {
     }
     case TSDB_DATA_TYPE_TIMESTAMP: {
       assert(0);
-      pVar->i64 = taosGetTimestamp(TSDB_TIME_PRECISION_NANO);                           
+      pVar->i = taosGetTimestamp(TSDB_TIME_PRECISION_NANO);                           
       break;                             
     }                            
     
@@ -151,43 +151,43 @@ void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uin
     case TSDB_DATA_TYPE_BOOL:
     case TSDB_DATA_TYPE_TINYINT: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->i64 = GET_INT8_VAL(pz);
+      pVar->i = GET_INT8_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_UTINYINT: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->u64 = GET_UINT8_VAL(pz);
+      pVar->u = GET_UINT8_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_SMALLINT: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->i64 = GET_INT16_VAL(pz);
+      pVar->i = GET_INT16_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_USMALLINT: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->u64 = GET_UINT16_VAL(pz);
+      pVar->u = GET_UINT16_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_INT: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->i64 = GET_INT32_VAL(pz);
+      pVar->i = GET_INT32_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_UINT: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->u64 = GET_UINT32_VAL(pz);
+      pVar->u = GET_UINT32_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_BIGINT:
     case TSDB_DATA_TYPE_TIMESTAMP: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->i64 = GET_INT64_VAL(pz);
+      pVar->i = GET_INT64_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_UBIGINT: {
       pVar->nLen = tDataTypes[type].bytes;
-      pVar->u64 = GET_UINT64_VAL(pz);
+      pVar->u = GET_UINT64_VAL(pz);
       break;
     }
     case TSDB_DATA_TYPE_DOUBLE: {
@@ -217,7 +217,7 @@ void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uin
     }
     
     default:
-      pVar->i64 = GET_INT32_VAL(pz);
+      pVar->i = GET_INT32_VAL(pz);
       pVar->nLen = tDataTypes[TSDB_DATA_TYPE_INT].bytes;
   }
   
@@ -271,7 +271,7 @@ void taosVariantAssign(SVariant *pDst, const SVariant *pSrc) {
   }
 
   if (IS_NUMERIC_TYPE(pSrc->nType) || (pSrc->nType == TSDB_DATA_TYPE_BOOL)) {
-    pDst->i64 = pSrc->i64;
+    pDst->i = pSrc->i;
   } else if (pSrc->nType == TSDB_DATA_TYPE_POINTER_ARRAY) {  // this is only for string array
     size_t num = taosArrayGetSize(pSrc->arr);
     pDst->arr = taosArrayInit(num, sizeof(char*));
@@ -322,16 +322,16 @@ int32_t taosVariantCompare(const SVariant* p1, const SVariant* p2) {
       return p1->d > p2->d? 1:-1;
     }
   } else if (IS_UNSIGNED_NUMERIC_TYPE(p1->nType)) {
-    if (p1->u64 == p2->u64) {
+    if (p1->u == p2->u) {
       return 0;
     } else {
-      return p1->u64 > p2->u64? 1:-1;
+      return p1->u > p2->u? 1:-1;
     }
   } else {
-    if (p1->i64 == p2->i64) {
+    if (p1->i == p2->i) {
       return 0;
     } else {
-      return p1->i64 > p2->i64? 1:-1;
+      return p1->i > p2->i? 1:-1;
     }
   }
 }
@@ -362,12 +362,12 @@ int32_t taosVariantToString(SVariant *pVar, char *dst) {
     case TSDB_DATA_TYPE_UTINYINT:
     case TSDB_DATA_TYPE_USMALLINT:
     case TSDB_DATA_TYPE_UINT:
-      return sprintf(dst, "%d", (int32_t)pVar->i64);
+      return sprintf(dst, "%d", (int32_t)pVar->i);
     
     case TSDB_DATA_TYPE_BIGINT:
-      return sprintf(dst, "%" PRId64, pVar->i64);
+      return sprintf(dst, "%" PRId64, pVar->i);
     case TSDB_DATA_TYPE_UBIGINT:
-      return sprintf(dst, "%" PRIu64, pVar->u64);
+      return sprintf(dst, "%" PRIu64, pVar->u);
     case TSDB_DATA_TYPE_FLOAT:
     case TSDB_DATA_TYPE_DOUBLE:
       return sprintf(dst, "%.9lf", pVar->d);
@@ -426,11 +426,11 @@ static int32_t toBinary(SVariant *pVariant, char **pDest, int32_t *pDestSize) {
     
   } else {
     if (IS_SIGNED_NUMERIC_TYPE(pVariant->nType)) {
-      sprintf(pBuf == NULL ? *pDest : pBuf, "%" PRId64, pVariant->i64);
+      sprintf(pBuf == NULL ? *pDest : pBuf, "%" PRId64, pVariant->i);
     } else if (pVariant->nType == TSDB_DATA_TYPE_DOUBLE || pVariant->nType == TSDB_DATA_TYPE_FLOAT) {
       sprintf(pBuf == NULL ? *pDest : pBuf, "%lf", pVariant->d);
     } else if (pVariant->nType == TSDB_DATA_TYPE_BOOL) {
-      sprintf(pBuf == NULL ? *pDest : pBuf, "%s", (pVariant->i64 == TSDB_TRUE) ? "TRUE" : "FALSE");
+      sprintf(pBuf == NULL ? *pDest : pBuf, "%s", (pVariant->i == TSDB_TRUE) ? "TRUE" : "FALSE");
     } else if (pVariant->nType == 0) {  // null data
       setNull(pBuf == NULL ? *pDest : pBuf, TSDB_DATA_TYPE_BINARY, 0);
     }
@@ -452,16 +452,16 @@ static int32_t toNchar(SVariant *pVariant, char **pDest, int32_t *pDestSize) {
 
   // convert the number to string, than convert it to wchar string.
   if (IS_SIGNED_NUMERIC_TYPE(pVariant->nType)) {
-    nLen = sprintf(pDst, "%" PRId64, pVariant->i64);
+    nLen = sprintf(pDst, "%" PRId64, pVariant->i);
   } else if (IS_UNSIGNED_NUMERIC_TYPE(pVariant->nType)) {
-    nLen = sprintf(pDst, "%"PRIu64, pVariant->u64);
+    nLen = sprintf(pDst, "%"PRIu64, pVariant->u);
   } else if (pVariant->nType == TSDB_DATA_TYPE_DOUBLE || pVariant->nType == TSDB_DATA_TYPE_FLOAT) {
     nLen = sprintf(pDst, "%lf", pVariant->d);
   } else if (pVariant->nType == TSDB_DATA_TYPE_BINARY) {
     pDst = pVariant->pz;
     nLen = pVariant->nLen;
   } else if (pVariant->nType == TSDB_DATA_TYPE_BOOL) {
-    nLen = sprintf(pDst, "%s", (pVariant->i64 == TSDB_TRUE) ? "TRUE" : "FALSE");
+    nLen = sprintf(pDst, "%s", (pVariant->i == TSDB_TRUE) ? "TRUE" : "FALSE");
   }
   
   if (*pDest == pVariant->pz) {
@@ -519,9 +519,9 @@ static FORCE_INLINE int32_t convertToInteger(SVariant *pVariant, int64_t *result
 #if 0
   errno = 0;
   if (IS_SIGNED_NUMERIC_TYPE(pVariant->nType) || (pVariant->nType == TSDB_DATA_TYPE_BOOL)) {
-    *result = pVariant->i64;
+    *result = pVariant->i;
   } else if (IS_UNSIGNED_NUMERIC_TYPE(pVariant->nType)) {
-    *result = pVariant->u64;
+    *result = pVariant->u;
   } else if (IS_FLOAT_TYPE(pVariant->nType)) {
     *result = (int64_t) pVariant->d;
   } else if (pVariant->nType == TSDB_DATA_TYPE_BINARY) {
@@ -639,9 +639,9 @@ static FORCE_INLINE int32_t convertToInteger(SVariant *pVariant, int64_t *result
 
 static int32_t convertToBool(SVariant *pVariant, int64_t *pDest) {
   if (pVariant->nType == TSDB_DATA_TYPE_BOOL) {
-    *pDest = pVariant->i64;  // in order to be compatible to null of bool
+    *pDest = pVariant->i;  // in order to be compatible to null of bool
   } else if (IS_NUMERIC_TYPE(pVariant->nType)) {
-    *pDest = ((pVariant->i64 != 0) ? TSDB_TRUE : TSDB_FALSE);
+    *pDest = ((pVariant->i != 0) ? TSDB_TRUE : TSDB_FALSE);
   } else if (pVariant->nType == TSDB_DATA_TYPE_FLOAT || pVariant->nType == TSDB_DATA_TYPE_DOUBLE) {
     *pDest = ((pVariant->d != 0) ? TSDB_TRUE : TSDB_FALSE);
   } else if (pVariant->nType == TSDB_DATA_TYPE_BINARY) {
@@ -791,12 +791,12 @@ int32_t tVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool inc
           *converted = true;
         }
         
-        if (pVariant->i64 > FLT_MAX || pVariant->i64 < -FLT_MAX) {          
-          SET_EXT_INFO(converted, pVariant->i64, -FLT_MAX, FLT_MAX, extInfo);
+        if (pVariant->i > FLT_MAX || pVariant->i < -FLT_MAX) {          
+          SET_EXT_INFO(converted, pVariant->i, -FLT_MAX, FLT_MAX, extInfo);
           return -1;
         }
 
-        SET_FLOAT_VAL(payload, pVariant->i64);
+        SET_FLOAT_VAL(payload, pVariant->i);
       } else if (IS_FLOAT_TYPE(pVariant->nType)) {
         if (converted) {
           *converted = true;
@@ -836,7 +836,7 @@ int32_t tVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool inc
           SET_DOUBLE_VAL(payload, value);
         }
       } else if (pVariant->nType == TSDB_DATA_TYPE_BOOL || IS_SIGNED_NUMERIC_TYPE(pVariant->nType) || IS_UNSIGNED_NUMERIC_TYPE(pVariant->nType)) {
-        SET_DOUBLE_VAL(payload, pVariant->i64);
+        SET_DOUBLE_VAL(payload, pVariant->i);
       } else if (IS_FLOAT_TYPE(pVariant->nType)) {
         SET_DOUBLE_VAL(payload, pVariant->d);
       } else if (pVariant->nType == TSDB_DATA_TYPE_NULL) {
@@ -885,7 +885,7 @@ int32_t tVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool inc
       if (pVariant->nType == TSDB_DATA_TYPE_NULL) {
         *((int64_t *)payload) = TSDB_DATA_BIGINT_NULL;
       } else {
-        *((int64_t *)payload) = pVariant->i64;
+        *((int64_t *)payload) = pVariant->i;
       }
       break;
     }
@@ -951,7 +951,7 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
   
   switch (type) {
     case TSDB_DATA_TYPE_BOOL: {  // bool
-      if (convertToBool(pVariant, &pVariant->i64) < 0) {
+      if (convertToBool(pVariant, &pVariant->i) < 0) {
         return -1;
       }
       
@@ -962,7 +962,7 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
     case TSDB_DATA_TYPE_BIGINT:
     case TSDB_DATA_TYPE_TINYINT:
     case TSDB_DATA_TYPE_SMALLINT: {
-      convertToInteger(pVariant, &(pVariant->i64), type, true, true, NULL);
+      convertToInteger(pVariant, &(pVariant->i), type, true, true, NULL);
       pVariant->nType = TSDB_DATA_TYPE_BIGINT;
       break;
     }
@@ -989,7 +989,7 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
         free(pVariant->pz);
         pVariant->d = v;
       } else if (pVariant->nType >= TSDB_DATA_TYPE_BOOL && pVariant->nType <= TSDB_DATA_TYPE_BIGINT) {
-        double tmp = (double) pVariant->i64;
+        double tmp = (double) pVariant->i;
         pVariant->d = tmp;
       }
       

source/libs/catalog/src/catalog.c

@@ -16,7 +16,7 @@
 #include "catalogInt.h"
 
 struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps) {
-  return NULL;
+  return (struct SCatalog*) 0x1;
 }
 
 int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData) {

source/libs/function/inc/texpr.h

@@ -62,7 +62,6 @@ typedef struct SExprTraverseSupp {
 
 tExprNode* exprTreeFromBinary(const void* data, size_t size);
 tExprNode* exprTreeFromTableName(const char* tbnameCond);
-tExprNode* exprdup(tExprNode* pTree);
 
 bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp *param);
 

source/libs/function/inc/tscalarfunction.h

@@ -28,6 +28,7 @@ extern struct SScalarFunctionInfo scalarFunc[1];
 #define FUNCTION_ROUND        40
 #define FUNCTION_MAVG         41
 #define FUNCTION_CSUM         42
+#define FUNCCTION_CONCAT      43
 
 #ifdef __cplusplus
 }

source/libs/function/src/texpr.c

@@ -228,7 +228,7 @@ void arithmeticTreeTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput,
 
     } else if (pRight->nodeType == TEXPR_VALUE_NODE) {  // exprLeft + 12
       _arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
-      OperatorFn(pLeftOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, &pRight->pVal->i64, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
+      OperatorFn(pLeftOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
     }
   } else if (pLeft->nodeType == TEXPR_COL_NODE) {
     // column data specified on left-hand-side
@@ -255,16 +255,16 @@ void arithmeticTreeTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput,
 
       if (order == TSDB_ORDER_DESC) {
         reverseCopy(pdata, pLeftInputData, pLeft->pSchema->type, numOfRows);
-        OperatorFn(pdata, numOfRows, pLeft->pSchema->type, &pRight->pVal->i64, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
+        OperatorFn(pdata, numOfRows, pLeft->pSchema->type, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
       } else {
-        OperatorFn(pLeftInputData, numOfRows, pLeft->pSchema->type, &pRight->pVal->i64, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
+        OperatorFn(pLeftInputData, numOfRows, pLeft->pSchema->type, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
       }
     }
   } else {
     // column data specified on left-hand-side
     if (pRight->nodeType == TEXPR_BINARYEXPR_NODE) {  // 12 + expr2
       _arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
-      OperatorFn(&pLeft->pVal->i64, 1, pLeft->pVal->nType, pRightOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pOutput, TSDB_ORDER_ASC);
+      OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, pRightOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pOutput, TSDB_ORDER_ASC);
 
     } else if (pRight->nodeType == TEXPR_COL_NODE) {  // 12 + columnRight
       // column data specified on right-hand-side
@@ -273,14 +273,14 @@ void arithmeticTreeTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput,
 
       if (order == TSDB_ORDER_DESC) {
         reverseCopy(pdata, pRightInputData, pRight->pSchema->type, numOfRows);
-        OperatorFn(&pLeft->pVal->i64, 1, pLeft->pVal->nType, pdata, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
+        OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, pdata, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
       } else {
-        OperatorFn(&pLeft->pVal->i64, 1, pLeft->pVal->nType, pRightInputData, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
+        OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, pRightInputData, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
       }
 
     } else if (pRight->nodeType == TEXPR_VALUE_NODE) {  // 12 + 12
       _arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
-      OperatorFn(&pLeft->pVal->i64, 1, pLeft->pVal->nType, &pRight->pVal->i64, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
+      OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
     }
   }
 
@@ -302,7 +302,7 @@ static void exprTreeToBinaryImpl(SBufferWriter* bw, tExprNode* expr) {
       tbufWriteInt32(bw, pVal->nLen);
       tbufWrite(bw, pVal->pz, pVal->nLen);
     } else {
-      tbufWriteInt64(bw, pVal->i64);
+      tbufWriteInt64(bw, pVal->i);
     }
     
   } else if (expr->nodeType == TEXPR_COL_NODE) {
@@ -371,7 +371,7 @@ static tExprNode* exprTreeFromBinaryImpl(SBufferReader* br) {
       pVal->pz = calloc(1, pVal->nLen + 1);
       tbufReadToBuffer(br, pVal->pz, pVal->nLen);
     } else {
-      pVal->i64 = tbufReadInt64(br);
+      pVal->i = tbufReadInt64(br);
     }
     
   } else if (pExpr->nodeType == TEXPR_COL_NODE) {

source/libs/function/src/ttszip.c

@@ -276,7 +276,7 @@ static void writeDataToDisk(STSBuf* pTSBuf) {
     metaLen += (int32_t)fwrite(&tfloat, 1, (size_t) pBlock->tag.nLen, pTSBuf->f);  
   } else if (pBlock->tag.nType != TSDB_DATA_TYPE_NULL) {
     metaLen += (int32_t)fwrite(&pBlock->tag.nLen, 1, sizeof(pBlock->tag.nLen), pTSBuf->f);
-    metaLen += (int32_t)fwrite(&pBlock->tag.i64, 1, (size_t) pBlock->tag.nLen, pTSBuf->f);
+    metaLen += (int32_t)fwrite(&pBlock->tag.i, 1, (size_t) pBlock->tag.nLen, pTSBuf->f);
   } else {
     trueLen = 0;
     metaLen += (int32_t)fwrite(&trueLen, 1, sizeof(pBlock->tag.nLen), pTSBuf->f);
@@ -364,7 +364,7 @@ STSBlock* readDataFromDisk(STSBuf* pTSBuf, int32_t order, bool decomp) {
     pBlock->tag.d = (double)tfloat;
     UNUSED(sz);
   } else if (pBlock->tag.nType != TSDB_DATA_TYPE_NULL) { //TODO check the return value
-    sz = fread(&pBlock->tag.i64, (size_t) pBlock->tag.nLen, 1, pTSBuf->f);
+    sz = fread(&pBlock->tag.i, (size_t) pBlock->tag.nLen, 1, pTSBuf->f);
     UNUSED(sz);
   }
 
@@ -984,7 +984,7 @@ void tsBufDisplay(STSBuf* pTSBuf) {
   while (tsBufNextPos(pTSBuf)) {
     STSElem elem = tsBufGetElem(pTSBuf);
     if (elem.tag->nType == TSDB_DATA_TYPE_BIGINT) {
-      printf("%d-%" PRId64 "-%" PRId64 "\n", elem.id, elem.tag->i64, elem.ts);
+      printf("%d-%" PRId64 "-%" PRId64 "\n", elem.id, elem.tag->i, elem.ts);
     }
   }
   

source/libs/parser/inc/queryInfoUtil.h

@@ -27,7 +27,6 @@ int32_t  getNumOfColumns(const STableMeta* pTableMeta);
 int32_t  getNumOfTags(const STableMeta* pTableMeta);
 SSchema *getTableColumnSchema(const STableMeta *pTableMeta);
 SSchema *getTableTagSchema(const STableMeta* pTableMeta);
-SSchema *getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex);
 
 size_t     getNumOfExprs(SQueryStmtInfo* pQueryInfo);
 SExprInfo* createBinaryExprInfo(struct tExprNode* pNode, SSchema* pResSchema);

source/libs/parser/src/astGenerator.c

@@ -375,25 +375,25 @@ void tSqlExprEvaluate(tSqlExpr* pExpr) {
 
       switch (optrType) {
         case TK_PLUS: {
-          pExpr->value.i64 = pLeft->value.i64 + pRight->value.i64;
+          pExpr->value.i = pLeft->value.i + pRight->value.i;
           break;
         }
         case TK_MINUS: {
-          pExpr->value.i64 = pLeft->value.i64 - pRight->value.i64;
+          pExpr->value.i = pLeft->value.i - pRight->value.i;
           break;
         }
         case TK_STAR: {
-          pExpr->value.i64 = pLeft->value.i64 * pRight->value.i64;
+          pExpr->value.i = pLeft->value.i * pRight->value.i;
           break;
         }
         case TK_DIVIDE: {
           pExpr->tokenId = TK_FLOAT;
           pExpr->value.nType = TSDB_DATA_TYPE_DOUBLE;
-          pExpr->value.d = (double)pLeft->value.i64 / pRight->value.i64;
+          pExpr->value.d = (double)pLeft->value.i / pRight->value.i;
           break;
         }
         case TK_REM: {
-          pExpr->value.i64 = pLeft->value.i64 % pRight->value.i64;
+          pExpr->value.i = pLeft->value.i % pRight->value.i;
           break;
         }
         default:
@@ -411,8 +411,8 @@ void tSqlExprEvaluate(tSqlExpr* pExpr) {
       pExpr->tokenId = TK_FLOAT;
       pExpr->type = SQL_NODE_VALUE;
 
-      double left  = (pLeft->value.nType  == TSDB_DATA_TYPE_DOUBLE) ? pLeft->value.d : pLeft->value.i64;
-      double right = (pRight->value.nType == TSDB_DATA_TYPE_DOUBLE) ? pRight->value.d : pRight->value.i64;
+      double left  = (pLeft->value.nType  == TSDB_DATA_TYPE_DOUBLE) ? pLeft->value.d : pLeft->value.i;
+      double right = (pRight->value.nType == TSDB_DATA_TYPE_DOUBLE) ? pRight->value.d : pRight->value.i;
 
       switch (optrType) {
         case TK_PLUS: {

source/libs/parser/src/parser.c

@@ -177,7 +177,7 @@ int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SMetaReq* pMet
       }
 
       // Let's assume that it is an UDF/UDAF, if it is not a built-in function.
-      if (!qIsBuiltinFunction(t->z, t->n)) {
+      if (qIsBuiltinFunction(t->z, t->n) < 0) {
         char* fname = strndup(t->z, t->n);
         taosArrayPush(pMetaInfo->pUdf, &fname);
       }

source/libs/parser/src/queryInfoUtil.c

@@ -212,7 +212,7 @@ void assignExprInfo(SExprInfo* dst, const SExprInfo* src) {
   }
 #endif
 
-//  dst->pExpr = exprdup(src->pExpr);
+  dst->pExpr = exprdup(src->pExpr);
   memset(dst->base.param, 0, sizeof(SVariant) * tListLen(dst->base.param));
   for (int32_t j = 0; j < src->base.numOfParams; ++j) {
     taosVariantAssign(&dst->base.param[j], &src->base.param[j]);

source/libs/parser/test/CMakeLists.txt

@@ -8,7 +8,7 @@ AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
 ADD_EXECUTABLE(parserTest ${SOURCE_LIST})
 TARGET_LINK_LIBRARIES(
         parserTest
-        PUBLIC os util common parser catalog transport gtest function
+        PUBLIC os util common parser catalog transport gtest function planner
 )
 
 TARGET_INCLUDE_DIRECTORIES(

source/libs/parser/test/plannerTest.cpp

+/*
+ * 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 <iostream>
+#pragma GCC diagnostic ignored "-Wwrite-strings"
+
+#pragma GCC diagnostic ignored "-Wunused-function"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+#include "os.h"
+
+#include "astGenerator.h"
+#include "parserInt.h"
+#include "taos.h"
+#include "tdef.h"
+#include "tvariant.h"
+#include "planner.h"
+
+namespace {
+void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_t colId) {
+  p->colId = colId;
+  p->bytes = bytes;
+  p->type = type;
+  strcpy(p->name, name);
+}
+
+void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SMetaReq *req) {
+  pQueryInfo->numOfTables = 1;
+
+  pQueryInfo->pTableMetaInfo = (STableMetaInfo**)calloc(1, POINTER_BYTES);
+  STableMetaInfo* pTableMetaInfo = (STableMetaInfo*)calloc(1, sizeof(STableMetaInfo));
+  pQueryInfo->pTableMetaInfo[0] = pTableMetaInfo;
+
+  SName* name = (SName*)taosArrayGet(req->pTableName, 0);
+
+  memcpy(&pTableMetaInfo->name, taosArrayGet(req->pTableName, 0), sizeof(SName));
+  pTableMetaInfo->pTableMeta = (STableMeta*)calloc(1, sizeof(STableMeta) + 4 * sizeof(SSchema));
+  strcpy(pTableMetaInfo->aliasName, name->tname);
+  STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
+  pTableMeta->tableType = TSDB_NORMAL_TABLE;
+  pTableMeta->tableInfo.numOfColumns = 4;
+  pTableMeta->tableInfo.rowSize = 28;
+  pTableMeta->uid = 110;
+
+  pTableMetaInfo->tagColList = (SArray*) taosArrayInit(4, POINTER_BYTES);
+
+  SSchema* pSchema = pTableMetaInfo->pTableMeta->schema;
+  setSchema(&pSchema[0], TSDB_DATA_TYPE_TIMESTAMP, 8, "ts", 0);
+  setSchema(&pSchema[1], TSDB_DATA_TYPE_INT, 4, "a", 1);
+  setSchema(&pSchema[2], TSDB_DATA_TYPE_DOUBLE, 8, "b", 2);
+  setSchema(&pSchema[3], TSDB_DATA_TYPE_DOUBLE, 8, "col", 3);
+
+}
+}
+
+TEST(testCase, planner_test) {
+  SSqlInfo info1 = doGenerateAST("select top(a*b / 99, 20) from `t.1abc` interval(10s, 1s)");
+  ASSERT_EQ(info1.valid, true);
+
+  char    msg[128] = {0};
+  SMsgBuf buf;
+  buf.len = 128;
+  buf.buf = msg;
+
+  SSqlNode* pNode = (SSqlNode*) taosArrayGetP(((SArray*)info1.list), 0);
+  int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
+  ASSERT_EQ(code, 0);
+
+  SMetaReq req = {0};
+  int32_t  ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
+  ASSERT_EQ(ret, 0);
+  ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
+
+  SQueryStmtInfo* pQueryInfo = createQueryInfo();
+  setTableMetaInfo(pQueryInfo, &req);
+
+  SSqlNode* pSqlNode = (SSqlNode*)taosArrayGetP(info1.list, 0);
+  ret = validateSqlNode(pSqlNode, pQueryInfo, &buf);
+  ASSERT_EQ(ret, 0);
+
+  SArray* pExprList = pQueryInfo->exprList;
+  ASSERT_EQ(taosArrayGetSize(pExprList), 2);
+
+  SExprInfo* p1 = (SExprInfo*) taosArrayGetP(pExprList, 1);
+  ASSERT_EQ(p1->base.uid, 110);
+  ASSERT_EQ(p1->base.numOfParams, 1);
+  ASSERT_EQ(p1->base.resSchema.type, TSDB_DATA_TYPE_DOUBLE);
+  ASSERT_STRCASEEQ(p1->base.resSchema.name, "top(a*b / 99, 20)");
+  ASSERT_EQ(p1->base.colInfo.flag, TSDB_COL_NORMAL);
+  ASSERT_STRCASEEQ(p1->base.token, "top(a*b / 99, 20)");
+  ASSERT_EQ(p1->base.interBytes, 16);
+
+  ASSERT_EQ(p1->pExpr->nodeType, TEXPR_UNARYEXPR_NODE);
+  ASSERT_EQ(p1->pExpr->_node.functionId, FUNCTION_TOP);
+  ASSERT_TRUE(p1->pExpr->_node.pRight == NULL);
+
+  tExprNode* pParam = p1->pExpr->_node.pLeft;
+
+  ASSERT_EQ(pParam->nodeType, TEXPR_BINARYEXPR_NODE);
+  ASSERT_EQ(pParam->_node.optr, TSDB_BINARY_OP_DIVIDE);
+  ASSERT_EQ(pParam->_node.pLeft->nodeType, TEXPR_BINARYEXPR_NODE);
+  ASSERT_EQ(pParam->_node.pRight->nodeType, TEXPR_VALUE_NODE);
+
+  ASSERT_EQ(taosArrayGetSize(pQueryInfo->colList), 3);
+  ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2);
+
+  struct SQueryPlanNode* n = nullptr;
+  code = qCreateQueryPlan(pQueryInfo, &n);
+
+  char* str = NULL;
+  qQueryPlanToString(n, &str);
+  printf("%s\n", str);
+
+  destroyQueryInfo(pQueryInfo);
+  qParserClearupMetaRequestInfo(&req);
+  destroySqlInfo(&info1);
+}
\ No newline at end of file

source/libs/parser/test/tokenizerTest.cpp

@@ -37,22 +37,22 @@ SToken createToken(char* s) {
 }  // namespace
 
 static void _init_tvariant_bool(SVariant* t) {
-  t->i64 = TSDB_FALSE;
+  t->i = TSDB_FALSE;
   t->nType = TSDB_DATA_TYPE_BOOL;
 }
 
 static void _init_tvariant_tinyint(SVariant* t) {
-  t->i64 = -27;
+  t->i = -27;
   t->nType = TSDB_DATA_TYPE_TINYINT;
 }
 
 static void _init_tvariant_int(SVariant* t) {
-  t->i64 = -23997659;
+  t->i = -23997659;
   t->nType = TSDB_DATA_TYPE_INT;
 }
 
 static void _init_tvariant_bigint(SVariant* t) {
-  t->i64 = -3333333333333;
+  t->i = -3333333333333;
   t->nType = TSDB_DATA_TYPE_BIGINT;
 }
 
@@ -359,19 +359,19 @@ TEST(testCase, tvariant_convert) {
   _init_tvariant_bool(&t);
 
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
-  EXPECT_EQ(t.i64, 0);
+  EXPECT_EQ(t.i, 0);
 
   _init_tvariant_bool(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_TINYINT), 0);
-  EXPECT_EQ(t.i64, 0);
+  EXPECT_EQ(t.i, 0);
 
   _init_tvariant_bool(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_SMALLINT), 0);
-  EXPECT_EQ(t.i64, 0);
+  EXPECT_EQ(t.i, 0);
 
   _init_tvariant_bool(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BIGINT), 0);
-  EXPECT_EQ(t.i64, 0);
+  EXPECT_EQ(t.i, 0);
 
   _init_tvariant_bool(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_FLOAT), 0);
@@ -394,23 +394,23 @@ TEST(testCase, tvariant_convert) {
   // 2. tinyint to other data types
   _init_tvariant_tinyint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
-  EXPECT_EQ(t.i64, 1);
+  EXPECT_EQ(t.i, 1);
 
   _init_tvariant_tinyint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_TINYINT), 0);
-  EXPECT_EQ(t.i64, -27);
+  EXPECT_EQ(t.i, -27);
 
   _init_tvariant_tinyint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_SMALLINT), 0);
-  EXPECT_EQ(t.i64, -27);
+  EXPECT_EQ(t.i, -27);
 
   _init_tvariant_tinyint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_INT), 0);
-  EXPECT_EQ(t.i64, -27);
+  EXPECT_EQ(t.i, -27);
 
   _init_tvariant_tinyint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BIGINT), 0);
-  EXPECT_EQ(t.i64, -27);
+  EXPECT_EQ(t.i, -27);
 
   _init_tvariant_tinyint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_FLOAT), 0);
@@ -434,7 +434,7 @@ TEST(testCase, tvariant_convert) {
   // types//////////////////////////////////////////////////////////////////
   _init_tvariant_int(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
-  EXPECT_EQ(t.i64, 1);
+  EXPECT_EQ(t.i, 1);
 
   _init_tvariant_int(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_TINYINT), 0);
@@ -444,11 +444,11 @@ TEST(testCase, tvariant_convert) {
 
   _init_tvariant_int(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_INT), 0);
-  EXPECT_EQ(t.i64, -23997659);
+  EXPECT_EQ(t.i, -23997659);
 
   _init_tvariant_int(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BIGINT), 0);
-  EXPECT_EQ(t.i64, -23997659);
+  EXPECT_EQ(t.i, -23997659);
 
   _init_tvariant_int(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_FLOAT), 0);
@@ -472,7 +472,7 @@ TEST(testCase, tvariant_convert) {
   // type//////////////////////////////////////////////////////////////////////////////
   _init_tvariant_bigint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
-  EXPECT_EQ(t.i64, 1);
+  EXPECT_EQ(t.i, 1);
 
   _init_tvariant_bigint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_TINYINT), 0);
@@ -485,7 +485,7 @@ TEST(testCase, tvariant_convert) {
 
   _init_tvariant_bigint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BIGINT), 0);
-  EXPECT_EQ(t.i64, -3333333333333);
+  EXPECT_EQ(t.i, -3333333333333);
 
   _init_tvariant_bigint(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_FLOAT), 0);
@@ -509,11 +509,11 @@ TEST(testCase, tvariant_convert) {
   // types////////////////////////////////////////////////////////////////////////
   _init_tvariant_float(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
-  EXPECT_EQ(t.i64, 1);
+  EXPECT_EQ(t.i, 1);
 
   _init_tvariant_float(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BIGINT), 0);
-  EXPECT_EQ(t.i64, -8991212199);
+  EXPECT_EQ(t.i, -8991212199);
 
   _init_tvariant_float(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_FLOAT), 0);
@@ -539,14 +539,14 @@ TEST(testCase, tvariant_convert) {
   t.nLen = strlen(t.pz);
   t.nType = TSDB_DATA_TYPE_BINARY;
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
-  EXPECT_EQ(t.i64, 1);
+  EXPECT_EQ(t.i, 1);
 
   _init_tvariant_binary(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), -1);
 
   _init_tvariant_binary(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BIGINT), 0);
-  EXPECT_EQ(t.i64, 200000);
+  EXPECT_EQ(t.i, 200000);
 
   _init_tvariant_binary(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_FLOAT), 0);
@@ -572,14 +572,14 @@ TEST(testCase, tvariant_convert) {
   t.nLen = wcslen(t.wpz);
   t.nType = TSDB_DATA_TYPE_NCHAR;
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
-  EXPECT_EQ(t.i64, 0);
+  EXPECT_EQ(t.i, 0);
 
   _init_tvariant_nchar(&t);
   EXPECT_LE(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BOOL), 0);
 
   _init_tvariant_nchar(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_BIGINT), 0);
-  EXPECT_EQ(t.i64, -2000000);
+  EXPECT_EQ(t.i, -2000000);
 
   _init_tvariant_nchar(&t);
   EXPECT_EQ(taosVariantTypeSetType(&t, TSDB_DATA_TYPE_FLOAT), 0);
@@ -667,59 +667,59 @@ TEST(testCase, isValidNumber_test) {
   EXPECT_EQ(tGetNumericStringType(&t1), TK_FLOAT);
 }
 
-//TEST(testCase, generateAST_test) {
-//  SSqlInfo info = doGenerateAST("select * from t1 where ts < now");
-//  ASSERT_EQ(info.valid, true);
-//
-//  SSqlInfo info1 = doGenerateAST("select * from `t.1abc` where ts<now+2h  and col < 20+99");
-//  ASSERT_EQ(info1.valid, true);
-//
-//  char msg[128] = {0};
-//
-//  SMsgBuf msgBuf = {0};
-//  msgBuf.buf = msg;
-//  msgBuf.len = 128;
-//
-//  SSqlNode* pNode = (SSqlNode*) taosArrayGetP(((SArray*)info1.list), 0);
-//  int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &msgBuf);
-//  ASSERT_EQ(code, 0);
-//
-//  SSqlInfo info2 = doGenerateAST("select * from abc where ts<now+2");
-//  SSqlNode* pNode2 = (SSqlNode*) taosArrayGetP(((SArray*)info2.list), 0);
-//  code = evaluateSqlNode(pNode2, TSDB_TIME_PRECISION_MILLI, &msgBuf);
-//  ASSERT_NE(code, 0);
-//
-//  destroySqlInfo(&info);
-//  destroySqlInfo(&info1);
-//  destroySqlInfo(&info2);
-//}
-//
-//TEST(testCase, evaluateAST_test) {
-//  SSqlInfo info1 = doGenerateAST("select a, b+22 from `t.1abc` where ts<now+2h and `col` < 20 + 99");
-//  ASSERT_EQ(info1.valid, true);
-//
-//  char msg[128] = {0};
-//  SMsgBuf msgBuf = {0};
-//  msgBuf.buf = msg;
-//  msgBuf.len = 128;
-//
-//  SSqlNode* pNode = (SSqlNode*) taosArrayGetP(((SArray*)info1.list), 0);
-//  int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &msgBuf);
-//  ASSERT_EQ(code, 0);
-//  destroySqlInfo(&info1);
-//}
-//
-//TEST(testCase, extractMeta_test) {
-//  SSqlInfo info1 = doGenerateAST("select a, b+22 from `t.1abc` where ts<now+2h and `col` < 20 + 99");
-//  ASSERT_EQ(info1.valid, true);
-//
-//  char msg[128] = {0};
-//  SMetaReq req  = {0};
-//  int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
-//  ASSERT_EQ(ret, 0);
-//  ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
-//
-//  qParserClearupMetaRequestInfo(&req);
-//  destroySqlInfo(&info1);
-//}
+TEST(testCase, generateAST_test) {
+  SSqlInfo info = doGenerateAST("select * from t1 where ts < now");
+  ASSERT_EQ(info.valid, true);
+
+  SSqlInfo info1 = doGenerateAST("select * from `t.1abc` where ts<now+2h  and col < 20+99");
+  ASSERT_EQ(info1.valid, true);
+
+  char msg[128] = {0};
+
+  SMsgBuf msgBuf = {0};
+  msgBuf.buf = msg;
+  msgBuf.len = 128;
+
+  SSqlNode* pNode = (SSqlNode*) taosArrayGetP(((SArray*)info1.list), 0);
+  int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &msgBuf);
+  ASSERT_EQ(code, 0);
+
+  SSqlInfo info2 = doGenerateAST("select * from abc where ts<now+2");
+  SSqlNode* pNode2 = (SSqlNode*) taosArrayGetP(((SArray*)info2.list), 0);
+  code = evaluateSqlNode(pNode2, TSDB_TIME_PRECISION_MILLI, &msgBuf);
+  ASSERT_NE(code, 0);
+
+  destroySqlInfo(&info);
+  destroySqlInfo(&info1);
+  destroySqlInfo(&info2);
+}
+
+TEST(testCase, evaluateAST_test) {
+  SSqlInfo info1 = doGenerateAST("select a, b+22 from `t.1abc` where ts<now+2h and `col` < 20 + 99");
+  ASSERT_EQ(info1.valid, true);
+
+  char msg[128] = {0};
+  SMsgBuf msgBuf = {0};
+  msgBuf.buf = msg;
+  msgBuf.len = 128;
+
+  SSqlNode* pNode = (SSqlNode*) taosArrayGetP(((SArray*)info1.list), 0);
+  int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &msgBuf);
+  ASSERT_EQ(code, 0);
+  destroySqlInfo(&info1);
+}
+
+TEST(testCase, extractMeta_test) {
+  SSqlInfo info1 = doGenerateAST("select a, b+22 from `t.1abc` where ts<now+2h and `col` < 20 + 99");
+  ASSERT_EQ(info1.valid, true);
+
+  char msg[128] = {0};
+  SMetaReq req  = {0};
+  int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
+  ASSERT_EQ(ret, 0);
+  ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
+
+  qParserClearupMetaRequestInfo(&req);
+  destroySqlInfo(&info1);
+}
 

source/libs/planner/CMakeLists.txt

@@ -9,4 +9,6 @@ target_include_directories(
 target_link_libraries(
     planner
     PRIVATE os util common catalog parser transport function
-)
\ No newline at end of file
+)
+
+ADD_SUBDIRECTORY(test)
\ No newline at end of file

source/libs/planner/inc/plannerUtil.h

+/*
+ * 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/>.
+ */
+
+#ifndef TDENGINE_PLANNERUTIL_H
+#define TDENGINE_PLANNERUTIL_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "parser.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#endif  // TDENGINE_PLANNERUTIL_H

source/libs/planner/src/planner.c

@@ -46,23 +46,20 @@ typedef struct SJoinCond {
 static SArray* createQueryPlanImpl(SQueryStmtInfo* pQueryInfo);
 static void doDestroyQueryNode(SQueryPlanNode* pQueryNode);
 
-  int32_t qOptimizeQueryPlan(struct SQueryPlanNode* pQueryNode) {
+int32_t qOptimizeQueryPlan(struct SQueryPlanNode* pQueryNode) {
   return 0;
 }
 
-int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode* pQueryNode) {
+int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode) {
   SArray* upstream = createQueryPlanImpl((struct SQueryStmtInfo*) pQueryInfo);
   assert(taosArrayGetSize(upstream) == 1);
 
-  /*SQueryPlanNode* p = */taosArrayGetP(upstream, 0);
+  *pQueryNode = taosArrayGetP(upstream, 0);
+
   taosArrayDestroy(upstream);
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qQueryPlanToString(struct SQueryPlanNode* pQueryNode, char** str) {
-  return 0;
-}
-
 int32_t qQueryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql) {
   return 0;
 }
@@ -108,10 +105,10 @@ static SQueryPlanNode* createQueryNode(int32_t type, const char* name, SQueryPla
   }
 
   pNode->numOfOutput = numOfOutput;
-  pNode->pExpr = calloc(numOfOutput, sizeof(SExprInfo));
+  pNode->pExpr = taosArrayInit(numOfOutput, POINTER_BYTES);
+
   for(int32_t i = 0; i < numOfOutput; ++i) {
-    SExprInfo* pExprInfo = taosArrayGet(pNode->pExpr, i);
-    assignExprInfo(pExprInfo, pExpr[i]);
+    taosArrayPush(pNode->pExpr, &pExpr[i]);
   }
 
   pNode->pPrevNodes = taosArrayInit(4, POINTER_BYTES);
@@ -167,7 +164,7 @@ static SQueryPlanNode* doAddTableColumnNode(SQueryStmtInfo* pQueryInfo, STableMe
     int32_t     num = (int32_t) taosArrayGetSize(pExprs);
     SQueryPlanNode* pNode = createQueryNode(QNODE_TAGSCAN, "TableTagScan", NULL, 0, pExprs->pData, num, info, NULL);
 
-    if (pQueryInfo->distinct) {
+    if (pQueryInfo->info.distinct) {
       pNode = createQueryNode(QNODE_DISTINCT, "Distinct", &pNode, 1, pExprs->pData, num, info, NULL);
     }
 
@@ -184,22 +181,21 @@ static SQueryPlanNode* doAddTableColumnNode(SQueryStmtInfo* pQueryInfo, STableMe
     // table source column projection, generate the projection expr
     int32_t     numOfCols = (int32_t) taosArrayGetSize(tableCols);
     SExprInfo** pExpr = calloc(numOfCols, POINTER_BYTES);
-    SSchema*    pSchema = pTableMetaInfo->pTableMeta->schema;
 
     STableMetaInfo* pTableMetaInfo1 = getMetaInfo(pQueryInfo, 0);
-    SSchema resultSchema = *pSchema;
-    resultSchema.colId = getNewResColId();
 
     for (int32_t i = 0; i < numOfCols; ++i) {
       SColumn* pCol = taosArrayGetP(tableCols, i);
       SColumnIndex index = {.tableIndex = 0, .columnIndex = pCol->columnIndex};
 
+      SSchema* pSchema = getOneColumnSchema(pTableMetaInfo->pTableMeta, i);
+      SSchema resultSchema = *pSchema;
+
       SExprInfo* p = createExprInfo(pTableMetaInfo1, FUNCTION_PRJ, &index, NULL, &resultSchema, 0);
       pExpr[i] = p;
     }
 
     pNode = createQueryNode(QNODE_PROJECT, "Projection", &pNode, 1, pExpr, numOfCols, info, NULL);
-//    dropAllExprInfo(pExpr);
     tfree(pExpr);
   }
 
@@ -254,7 +250,7 @@ static SQueryPlanNode* doCreateQueryPlanForOneTableImpl(SQueryStmtInfo* pQueryIn
 static SQueryPlanNode* doCreateQueryPlanForOneTable(SQueryStmtInfo* pQueryInfo, STableMetaInfo* pTableMetaInfo, SArray* pExprs,
                                                 SArray* tableCols) {
   char name[TSDB_TABLE_FNAME_LEN] = {0};
-  tNameExtractFullName(&pTableMetaInfo->name, name);
+  tstrncpy(name, pTableMetaInfo->name.tname, TSDB_TABLE_FNAME_LEN);
 
   SQueryTableInfo info = {.tableName = strdup(name), .uid = pTableMetaInfo->pTableMeta->uid,};
 
@@ -357,12 +353,6 @@ static void doDestroyQueryNode(SQueryPlanNode* pQueryNode) {
   tfree(pQueryNode);
 }
 
-bool hasAliasName(SExprInfo* pExpr) {
-  assert(pExpr != NULL);
-  return true;
-//  return strncmp(pExpr->base.token, pExpr->base., tListLen(pExpr->base.aliasName)) != 0;
-}
-
 static int32_t doPrintPlan(char* buf, SQueryPlanNode* pQueryNode, int32_t level, int32_t totalLen) {
   if (level > 0) {
     sprintf(buf + totalLen, "%*c", level, ' ');
@@ -412,13 +402,7 @@ static int32_t doPrintPlan(char* buf, SQueryPlanNode* pQueryNode, int32_t level,
         SExprInfo* pExprInfo = taosArrayGetP(pQueryNode->pExpr, i);
 
         SSqlExpr* pExpr = &pExprInfo->base;
-//        if (hasAliasName(&pQueryNode->pExpr[i])) {
-          len1 = sprintf(buf + len,"[%s #%s]", pExpr->token, pExpr->resSchema.name);
-//        } else {
-//          len1 = sprintf(buf + len,"[%s]", pExpr->token);
-//        }
-
-        len += len1;
+        len += sprintf(buf + len,"%s [%s #%d]", pExpr->token, pExpr->resSchema.name, pExpr->resSchema.colId);
         if (i < pQueryNode->numOfOutput - 1) {
           len1 = sprintf(buf + len, ", ");
           len += len1;
@@ -435,13 +419,7 @@ static int32_t doPrintPlan(char* buf, SQueryPlanNode* pQueryNode, int32_t level,
         SExprInfo* pExprInfo = taosArrayGetP(pQueryNode->pExpr, i);
 
         SSqlExpr* pExpr = &pExprInfo->base;
-        if (hasAliasName(pExprInfo)) {
-          len1 = sprintf(buf + len,"[%s #%s]", pExpr->token, pExpr->resSchema.name);
-        } else {
-          len1 = sprintf(buf + len,"[%s]", pExpr->token);
-        }
-
-        len += len1;
+        len += sprintf(buf + len,"%s [%s #%d]", pExpr->token, pExpr->resSchema.name, pExpr->resSchema.colId);
         if (i < pQueryNode->numOfOutput - 1) {
           len1 = sprintf(buf + len,", ");
           len += len1;
@@ -452,9 +430,11 @@ static int32_t doPrintPlan(char* buf, SQueryPlanNode* pQueryNode, int32_t level,
       len += len1;
 
       SInterval* pInterval = pQueryNode->pExtInfo;
-      len1 = sprintf(buf + len, "interval:%" PRId64 "(%s), sliding:%" PRId64 "(%s), offset:%" PRId64 "\n",
+
+      // todo dynamic return the time precision
+      len1 = sprintf(buf + len, "interval:%" PRId64 "(%s), sliding:%" PRId64 "(%s), offset:%" PRId64 "(%s)\n",
                      pInterval->interval, TSDB_TIME_PRECISION_MILLI_STR, pInterval->sliding, TSDB_TIME_PRECISION_MILLI_STR,
-                     pInterval->offset);
+                     pInterval->offset, TSDB_TIME_PRECISION_MILLI_STR);
       len += len1;
 
       break;
@@ -465,12 +445,7 @@ static int32_t doPrintPlan(char* buf, SQueryPlanNode* pQueryNode, int32_t level,
         SExprInfo* pExprInfo = taosArrayGetP(pQueryNode->pExpr, i);
 
         SSqlExpr* pExpr = &pExprInfo->base;
-
-        if (hasAliasName(pExprInfo)) {
-          len1 = sprintf(buf + len,"[%s #%s]", pExpr->token, pExpr->resSchema.name);
-        } else {
-          len1 = sprintf(buf + len,"[%s]", pExpr->token);
-        }
+        len1 = sprintf(buf + len,"%s [%s #%d]", pExpr->token, pExpr->resSchema.name, pExpr->resSchema.colId);
 
         len += len1;
         if (i < pQueryNode->numOfOutput - 1) {
@@ -568,14 +543,15 @@ int32_t queryPlanToStringImpl(char* buf, SQueryPlanNode* pQueryNode, int32_t lev
   return len;
 }
 
-char* queryPlanToString(SQueryPlanNode* pQueryNode) {
+int32_t qQueryPlanToString(struct SQueryPlanNode* pQueryNode, char** str) {
   assert(pQueryNode);
 
-  char* buf = calloc(1, 4096);
+  *str = calloc(1, 4096);
 
-  int32_t len = sprintf(buf, "===== logic plan =====\n");
-  queryPlanToStringImpl(buf, pQueryNode, 0, len);
-  return buf;
+  int32_t len = sprintf(*str, "===== logic plan =====\n");
+  queryPlanToStringImpl(*str, pQueryNode, 0, len);
+
+  return TSDB_CODE_SUCCESS;
 }
 
 SQueryPlanNode* queryPlanFromString() {

source/libs/planner/src/plannerUtil.c

+#include "tvariant.h"
+#include "plannerUtil.h"
+
+
+
+

source/libs/planner/test/CMakeLists.txt

+
+MESSAGE(STATUS "build planner unit test")
+
+# GoogleTest requires at least C++11
+SET(CMAKE_CXX_STANDARD 11)
+AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
+
+ADD_EXECUTABLE(plannerTest ${SOURCE_LIST})
+TARGET_LINK_LIBRARIES(
+        plannerTest
+        PUBLIC os util common planner parser catalog transport gtest function
+)
+
+TARGET_INCLUDE_DIRECTORIES(
+        plannerTest
+        PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/planner/"
+        PRIVATE "${CMAKE_SOURCE_DIR}/source/libs/planner/inc"
+)

source/libs/planner/test/plannerTests.cpp

+/*
+ * 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 <iostream>
+
+#include "os.h"
+
+#include "taos.h"
+#include "parser.h"
+
+#pragma GCC diagnostic ignored "-Wwrite-strings"
+
+#pragma GCC diagnostic ignored "-Wunused-function"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+
+int main(int argc, char** argv) {
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+TEST(testCase, planner_test) {
+  char    msg[128] = {0};
+  const char* sql = "select top(a*b / 99, 20) from `t.1abc` interval(10s, 1s)";
+
+  SQueryStmtInfo* pQueryInfo = nullptr;
+//  int32_t code = qParseQuerySql(sql, strlen(sql), &pQueryInfo, 0, msg, sizeof(msg));
+//  ASSERT_EQ(code, 0);
+
+//  SSqlNode* pNode = (SSqlNode*)taosArrayGetP(((SArray*)info1.list), 0);
+//  int32_t   code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
+//  ASSERT_EQ(code, 0);
+//
+//  SMetaReq req = {0};
+//  int32_t  ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
+//  ASSERT_EQ(ret, 0);
+//  ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
+//
+//  SQueryStmtInfo* pQueryInfo = createQueryInfo();
+//  setTableMetaInfo(pQueryInfo, &req);
+//
+//  SSqlNode* pSqlNode = (SSqlNode*)taosArrayGetP(info1.list, 0);
+//  ret = validateSqlNode(pSqlNode, pQueryInfo, &buf);
+//  ASSERT_EQ(ret, 0);
+//
+//  SArray* pExprList = pQueryInfo->exprList;
+//  ASSERT_EQ(taosArrayGetSize(pExprList), 2);
+//
+//  SExprInfo* p1 = (SExprInfo*)taosArrayGetP(pExprList, 1);
+//  ASSERT_EQ(p1->base.uid, 110);
+//  ASSERT_EQ(p1->base.numOfParams, 1);
+//  ASSERT_EQ(p1->base.resSchema.type, TSDB_DATA_TYPE_DOUBLE);
+//  ASSERT_STRCASEEQ(p1->base.resSchema.name, "top(a*b / 99, 20)");
+//  ASSERT_EQ(p1->base.colInfo.flag, TSDB_COL_NORMAL);
+//  ASSERT_STRCASEEQ(p1->base.token, "top(a*b / 99, 20)");
+//  ASSERT_EQ(p1->base.interBytes, 16);
+//
+//  ASSERT_EQ(p1->pExpr->nodeType, TEXPR_UNARYEXPR_NODE);
+//  ASSERT_EQ(p1->pExpr->_node.functionId, FUNCTION_TOP);
+//  ASSERT_TRUE(p1->pExpr->_node.pRight == NULL);
+//
+//  tExprNode* pParam = p1->pExpr->_node.pLeft;
+//
+//  ASSERT_EQ(pParam->nodeType, TEXPR_BINARYEXPR_NODE);
+//  ASSERT_EQ(pParam->_node.optr, TSDB_BINARY_OP_DIVIDE);
+//  ASSERT_EQ(pParam->_node.pLeft->nodeType, TEXPR_BINARYEXPR_NODE);
+//  ASSERT_EQ(pParam->_node.pRight->nodeType, TEXPR_VALUE_NODE);
+//
+//  ASSERT_EQ(taosArrayGetSize(pQueryInfo->colList), 3);
+//  ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2);
+//
+//  destroyQueryInfo(pQueryInfo);
+//  qParserClearupMetaRequestInfo(&req);
+//  destroySqlInfo(&info1);
+}
\ No newline at end of file

source/libs/wal/CMakeLists.txt

@@ -4,9 +4,9 @@ target_include_directories(
     wal
     PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/wal"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
 )
 
 target_link_libraries(
-    os
+    wal
+    PUBLIC os
 )

source/server/vnode/CMakeLists.txt

 add_subdirectory(meta)
 add_subdirectory(tq)
 add_subdirectory(tsdb)
-
-aux_source_directory(src VNODE_SRC)
-add_library(vnode STATIC ${VNODE_SRC})
-target_include_directories(
-    vnode
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/server/vnode"
-    private "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-target_link_libraries(
-    vnode
-    PUBLIC transport
-    PUBLIC meta
-    PUBLIC tq
-    PUBLIC tsdb
-    PUBLIC wal
-    PUBLIC sync
-    PUBLIC cjson
-)
-
-# test
-if(${BUILD_TEST})
-    add_subdirectory(test)
-endif(${BUILD_TEST})
\ No newline at end of file
+add_subdirectory(impl)
\ No newline at end of file

source/server/vnode/impl/CMakeLists.txt

+aux_source_directory(src VNODE_SRC)
+add_library(vnode STATIC ${VNODE_SRC})
+target_include_directories(
+    vnode
+    PUBLIC "${CMAKE_SOURCE_DIR}/include/server/vnode"
+    private "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+target_link_libraries(
+    vnode
+    PUBLIC transport
+    PUBLIC meta
+    PUBLIC tq
+    PUBLIC tsdb
+    PUBLIC wal
+    PUBLIC sync
+    PUBLIC cjson
+)
+
+# test
+if(${BUILD_TEST})
+    add_subdirectory(test)
+endif(${BUILD_TEST})
\ No newline at end of file

source/server/vnode/src/vnodeReadMsg.c → source/server/vnode/impl/src/vnodeReadMsg.c

@@ -217,8 +217,8 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SReadMsg *pRead) {
 int32_t vnodeProcessConsumeMsg(SVnode *pVnode, SReadMsg *pRead) {
   //parse message and optionally move offset
   void* pMsg = pRead->pCont;
-  tmqConsumeReq *pConsumeMsg = (tmqConsumeReq*) pMsg;
-  tmqMsgHead msgHead = pConsumeMsg->head;
+  TmqConsumeReq *pConsumeMsg = (TmqConsumeReq*) pMsg;
+  TmqMsgHead msgHead = pConsumeMsg->head;
   //extract head
   STQ *pTq = pVnode->pTQ;
   /*tqBufferHandle *pHandle = tqGetHandle(pTq, msgHead.clientId);*/

source/server/vnode/meta/src/metaMain.c

@@ -64,6 +64,7 @@ SMeta *metaOpen(SMetaOpts *pMetaOpts) {
 
   // TODO: need to figure out how to persist the START UID
   tableUidGeneratorInit(&(pMeta->uidGenerator), IVLD_TB_UID);
+  return pMeta;
 }
 
 void metaClose(SMeta *pMeta) {

source/server/vnode/tq/CMakeLists.txt

@@ -3,11 +3,12 @@ add_library(tq ${TQ_SRC})
 target_include_directories(
     tq
     PUBLIC "${CMAKE_SOURCE_DIR}/include/server/vnode/tq"
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/wal"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
 )
 
 target_link_libraries(
-    wal
+    tq
+    PUBLIC wal
+    PUBLIC os
+    PUBLIC util
 )

source/server/vnode/tq/inc/tqCommit.h

+/*
+ * 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/>.
+ */

source/server/vnode/tq/inc/tqInt.h

@@ -18,7 +18,6 @@
 
 #include "tq.h"
 
-
 #ifdef __cplusplus
 extern "C" {
 #endif

source/server/vnode/tq/inc/tqMetaStore.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TQ_META_STORE_H_
+#define _TQ_META_STORE_H_
+
+#include "os.h"
+
+#define TQ_INUSE_SIZE 0xFF
+#define TQ_PAGE_SIZE 4096
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct TqMetaHandle {
+  int64_t key;
+  int64_t offset;
+  void *valueInUse;
+  void *valueInTxn;
+} TqMetaHandle;
+
+typedef struct TqMetaList {
+  TqMetaHandle handle;
+  struct TqMetaList* next;
+  struct TqMetaList* inTxnPrev;
+  struct TqMetaList* inTxnNext;
+  struct TqMetaList* unpersistPrev;
+  struct TqMetaList* unpersistNext;
+} TqMetaList;
+
+typedef struct TqMetaStore {
+  TqMetaList* inUse[TQ_INUSE_SIZE];
+  //a table head, key is empty
+  TqMetaList* unpersistHead;
+  int fileFd; //TODO:temporaral use
+  int idxFd;  //TODO:temporaral use
+  void* (*serializer)(void*);
+  void* (*deserializer)(void*);
+  void  (*deleter)(void*);
+} TqMetaStore;
+
+TqMetaStore*  tqStoreOpen(const char* path, void* serializer(void* ), void* deserializer(void*), void deleter(void*));
+int32_t       tqStoreClose(TqMetaStore*);
+int32_t       tqStoreDelete(TqMetaStore*);
+//int32_t       TqStoreCommitAll(TqMetaStore*);
+int32_t       tqStorePersist(TqMetaStore*);
+
+TqMetaHandle* tqHandleGetInUse(TqMetaStore*, int64_t key);
+int32_t       tqHandlePutInUse(TqMetaStore*, TqMetaHandle* handle);
+TqMetaHandle* tqHandleGetInTxn(TqMetaStore*, int64_t key);
+int32_t       tqHandlePutInTxn(TqMetaStore*, TqMetaHandle* handle);
+//delete in-use-handle, make in-txn-handle in use
+int32_t       tqHandleCommit(TqMetaStore*, int64_t key);
+//delete in-txn-handle
+int32_t       tqHandleAbort(TqMetaStore*, int64_t key);
+//delete in-use-handle
+int32_t       tqHandleDel(TqMetaStore*, int64_t key);
+//delete in-use-handle and in-txn-handle
+int32_t       tqHandleClear(TqMetaStore*, int64_t key);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* ifndef _TQ_META_STORE_H_ */

source/server/vnode/tq/src/tq.c

@@ -22,18 +22,18 @@
 //
 //handle management message
 //
-static int tqProtoCheck(tmqMsgHead *pMsg) {
+static int tqProtoCheck(TmqMsgHead *pMsg) {
   return pMsg->protoVer == 0;
 }
 
-static int tqAckOneTopic(tqBufferHandle *bhandle, tmqOneAck *pAck, tqQueryMsg** ppQuery) {
+static int tqAckOneTopic(TqBufferHandle *bhandle, TmqOneAck *pAck, TqQueryMsg** ppQuery) {
   //clean old item and move forward
   int32_t consumeOffset = pAck->consumeOffset;
   int idx = consumeOffset % TQ_BUFFER_SIZE;
   ASSERT(bhandle->buffer[idx].content && bhandle->buffer[idx].executor);
   tfree(bhandle->buffer[idx].content);
   if( 1 /* TODO: need to launch new query */) {
-    tqQueryMsg* pNewQuery = malloc(sizeof(tqQueryMsg));
+    TqQueryMsg* pNewQuery = malloc(sizeof(TqQueryMsg));
     if(pNewQuery == NULL) {
       //TODO: memory insufficient
       return -1;
@@ -49,19 +49,19 @@ static int tqAckOneTopic(tqBufferHandle *bhandle, tmqOneAck *pAck, tqQueryMsg**
   return 0;
 }
 
-static int tqAck(tqGroupHandle* ghandle, tmqAcks* pAcks) {
+static int tqAck(TqGroupHandle* ghandle, TmqAcks* pAcks) {
   int32_t ackNum = pAcks->ackNum;
-  tmqOneAck *acks = pAcks->acks;
+  TmqOneAck *acks = pAcks->acks;
   //double ptr for acks and list
   int i = 0;
-  tqListHandle* node = ghandle->head;
+  TqListHandle* node = ghandle->head;
   int ackCnt = 0;
-  tqQueryMsg *pQuery = NULL;
+  TqQueryMsg *pQuery = NULL;
   while(i < ackNum && node->next) {
-    if(acks[i].topicId == node->next->bufHandle->topicId) {
+    if(acks[i].topicId == node->next->bufHandle.topicId) {
       ackCnt++;
-      tqAckOneTopic(node->next->bufHandle, &acks[i], &pQuery);
-    } else if(acks[i].topicId < node->next->bufHandle->topicId) {
+      tqAckOneTopic(&node->next->bufHandle, &acks[i], &pQuery);
+    } else if(acks[i].topicId < node->next->bufHandle.topicId) {
       i++;
     } else {
       node = node->next;
@@ -73,12 +73,12 @@ static int tqAck(tqGroupHandle* ghandle, tmqAcks* pAcks) {
   return ackCnt;
 }
 
-static int tqCommitTCGroup(tqGroupHandle* handle) {
+static int tqCommitTCGroup(TqGroupHandle* handle) {
   //persist modification into disk
   return 0;
 }
 
-int tqCreateTCGroup(STQ *pTq, int64_t topicId, int64_t cgId, int64_t cId, tqGroupHandle** handle) {
+int tqCreateTCGroup(STQ *pTq, int64_t topicId, int64_t cgId, int64_t cId, TqGroupHandle** handle) {
   //create in disk
   return 0;
 }
@@ -99,13 +99,13 @@ int tqDropTCGroup(STQ* pTq, int64_t topicId, int64_t cgId, int64_t cId) {
   return 0;
 }
 
-static int tqFetch(tqGroupHandle* ghandle, void** msg) {
-  tqListHandle* head = ghandle->head;
-  tqListHandle* node = head;
+static int tqFetch(TqGroupHandle* ghandle, void** msg) {
+  TqListHandle* head = ghandle->head;
+  TqListHandle* node = head;
   int totSize = 0;
   //TODO: make it a macro
   int sizeLimit = 4 * 1024;
-  tmqMsgContent* buffer = malloc(sizeLimit);
+  TmqMsgContent* buffer = malloc(sizeLimit);
   if(buffer == NULL) {
     //TODO:memory insufficient
     return -1;
@@ -114,7 +114,7 @@ static int tqFetch(tqGroupHandle* ghandle, void** msg) {
   //until all topic iterated or msgs over sizeLimit
   while(node->next) {
     node = node->next;
-    tqBufferHandle* bufHandle = node->bufHandle;
+    TqBufferHandle* bufHandle = &node->bufHandle;
     int idx = bufHandle->nextConsumeOffset % TQ_BUFFER_SIZE;
     if(bufHandle->buffer[idx].content != NULL &&
         bufHandle->buffer[idx].offset == bufHandle->nextConsumeOffset
@@ -140,28 +140,23 @@ static int tqFetch(tqGroupHandle* ghandle, void** msg) {
       }
     }
   }
-  if(totSize == 0) {
-    //no msg
-    return -1;
-  }
-
   return totSize;
 }
 
 
-tqGroupHandle* tqGetGroupHandle(STQ* pTq, int64_t cId) {
+TqGroupHandle* tqGetGroupHandle(STQ* pTq, int64_t cId) {
   return NULL;
 }
 
-int tqLaunchQuery(tqGroupHandle* ghandle) {
+int tqLaunchQuery(TqGroupHandle* ghandle) {
   return 0;
 }
 
-int tqSendLaunchQuery(STQ* pTq, int64_t topicId, int64_t cgId, void* query) {
+int tqSendLaunchQuery(TqGroupHandle* gHandle) {
   return 0;
 }
 
-/*int tqMoveOffsetToNext(tqGroupHandle* ghandle) {*/
+/*int tqMoveOffsetToNext(TqGroupHandle* ghandle) {*/
   /*return 0;*/
 /*}*/
 
@@ -176,13 +171,13 @@ int tqCommit(STQ* pTq) {
   return 0;
 }
 
-int tqConsume(STQ* pTq, tmqConsumeReq* pMsg) {
-  if(!tqProtoCheck((tmqMsgHead *)pMsg)) {
+int tqConsume(STQ* pTq, TmqConsumeReq* pMsg) {
+  if(!tqProtoCheck((TmqMsgHead *)pMsg)) {
     //proto version invalid
     return -1;
   }
   int64_t clientId = pMsg->head.clientId;
-  tqGroupHandle *ghandle = tqGetGroupHandle(pTq, clientId);
+  TqGroupHandle *ghandle = tqGetGroupHandle(pTq, clientId);
   if(ghandle == NULL) {
     //client not connect
     return -1;
@@ -194,9 +189,9 @@ int tqConsume(STQ* pTq, tmqConsumeReq* pMsg) {
     }
   }
 
-  tmqConsumeRsp *pRsp = (tmqConsumeRsp*) pMsg;
+  TmqConsumeRsp *pRsp = (TmqConsumeRsp*) pMsg;
 
-  if(tqFetch(ghandle, (void**)&pRsp->msgs) < 0) {
+  if(tqFetch(ghandle, (void**)&pRsp->msgs) <= 0) {
     //fetch error
     return -1;
   }
@@ -209,14 +204,9 @@ int tqConsume(STQ* pTq, tmqConsumeReq* pMsg) {
   return 0;
 }
 
-
-int tqSerializeGroupHandle(tqGroupHandle *gHandle, void** ppBytes, int32_t offset) {
+int tqSerializeGroupHandle(TqGroupHandle *gHandle, void** ppBytes) {
   //calculate size
   int sz = tqGetGHandleSSize(gHandle);
-  if(sz <= 0) {
-    //TODO: err
-    return -1;
-  }
   void* ptr = realloc(*ppBytes, sz);
   if(ptr == NULL) {
     free(ppBytes);
@@ -224,29 +214,30 @@ int tqSerializeGroupHandle(tqGroupHandle *gHandle, void** ppBytes, int32_t offse
     return -1;
   }
   *ppBytes = ptr;
-  //do serialize
+  //do serialization
   *(int64_t*)ptr = gHandle->cId;
   ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
   *(int64_t*)ptr = gHandle->cgId;
   ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
   *(int32_t*)ptr = gHandle->topicNum;
+  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
   if(gHandle->topicNum > 0) {
-    tqSerializeListHandle(gHandle->head, ppBytes, ptr - *ppBytes);
+    tqSerializeListHandle(gHandle->head, ptr);
   }
   return 0;
 }
 
-int tqSerializeListHandle(tqListHandle *listHandle, void** ppBytes, int32_t offset) {
-  void* ptr = POINTER_SHIFT(*ppBytes, offset);
-  tqListHandle *node = listHandle;
-  while(node->next) {
+void* tqSerializeListHandle(TqListHandle *listHandle, void* ptr) {
+  TqListHandle *node = listHandle;
+  ASSERT(node != NULL);
+  while(node) {
+    ptr = tqSerializeBufHandle(&node->bufHandle, ptr);
     node = node->next;
-    offset = tqSerializeBufHandle(node->bufHandle, ppBytes, offset);
   }
-  return offset;
+  return ptr;
 }
-int tqSerializeBufHandle(tqBufferHandle *bufHandle, void** ppBytes, int32_t offset) {
-  void *ptr = POINTER_SHIFT(*ppBytes, offset);
+
+void* tqSerializeBufHandle(TqBufferHandle *bufHandle, void* ptr) {
   *(int64_t*)ptr = bufHandle->nextConsumeOffset;
   ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
   *(int64_t*)ptr = bufHandle->topicId;
@@ -256,41 +247,87 @@ int tqSerializeBufHandle(tqBufferHandle *bufHandle, void** ppBytes, int32_t offs
   *(int32_t*)ptr = bufHandle->tail;
   ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
   for(int i = 0; i < TQ_BUFFER_SIZE; i++) {
-    int sz = tqSerializeBufItem(&bufHandle->buffer[i], ppBytes, ptr - *ppBytes);
-    ptr = POINTER_SHIFT(ptr, sz);
+    ptr = tqSerializeBufItem(&bufHandle->buffer[i], ptr);
   }
-  return ptr - *ppBytes;
+  return ptr;
 }
 
-int tqSerializeBufItem(tqBufferItem *bufItem, void** ppBytes, int32_t offset) {
-  void *ptr = POINTER_SHIFT(*ppBytes, offset);
+void* tqSerializeBufItem(TqBufferItem *bufItem, void* ptr) {
   //TODO: do we need serialize this? 
-  return 0;
+  //mainly for executor
+  return ptr;
 }
 
-int tqDeserializeGroupHandle(const void* pBytes, tqGroupHandle **pGhandle) {
-  return 0;
-}
-int tqDeserializeListHandle(const void* pBytes, tqListHandle **pListHandle) {
-  return 0;
-}
-int tqDeserializeBufHandle(const void* pBytes, tqBufferHandle **pBufHandle) {
-  return 0;
-}
-int tqDeserializeBufItem(const void* pBytes, tqBufferItem **pBufItem) {
-  return 0;
+const void* tqDeserializeGroupHandle(const void* pBytes, TqGroupHandle *gHandle) {
+  const void* ptr = pBytes;
+  gHandle->cId = *(int64_t*)ptr;
+  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  gHandle->cgId = *(int64_t*)ptr;
+  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  gHandle->ahandle = NULL;
+  gHandle->topicNum = *(int32_t*)ptr;
+  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  gHandle->head = NULL;
+  TqListHandle *node = gHandle->head;
+  for(int i = 0; i < gHandle->topicNum; i++) {
+    if(gHandle->head == NULL) {
+      if((node = malloc(sizeof(TqListHandle))) == NULL) {
+        //TODO: error
+        return NULL;
+      }
+      node->next= NULL;
+      ptr = tqDeserializeBufHandle(ptr, &node->bufHandle); 
+      gHandle->head = node;
+    } else {
+      node->next = malloc(sizeof(TqListHandle));
+      if(node->next == NULL) {
+        //TODO: error
+        return NULL;
+      }
+      node->next->next = NULL;
+      ptr = tqDeserializeBufHandle(ptr, &node->next->bufHandle);
+      node = node->next;
+    }
+  }
+  return ptr;
 }
 
+const void* tqDeserializeBufHandle(const void* pBytes, TqBufferHandle *bufHandle) {
+  const void* ptr = pBytes;
+  bufHandle->nextConsumeOffset = *(int64_t*)ptr;
+  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  bufHandle->topicId = *(int64_t*)ptr;
+  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  bufHandle->head = *(int32_t*)ptr;
+  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  bufHandle->tail = *(int32_t*)ptr;
+  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  for(int i = 0; i < TQ_BUFFER_SIZE; i++) {
+    ptr = tqDeserializeBufItem(ptr, &bufHandle->buffer[i]);
+  }
+  return ptr;
+}
 
-int tqGetGHandleSSize(const tqGroupHandle *gHandle) {
-  return 0;
+const void* tqDeserializeBufItem(const void* pBytes, TqBufferItem *bufItem) {
+  return pBytes;
 }
-int tqListHandleSSize(const tqListHandle *listHandle) {
-  return 0;
+
+//TODO: make this a macro
+int tqGetGHandleSSize(const TqGroupHandle *gHandle) {
+  return sizeof(int64_t) * 2
+    + sizeof(int32_t)
+    + gHandle->topicNum * tqBufHandleSSize();
 }
-int tqBufHandleSSize(const tqBufferHandle *bufHandle) {
-  return 0;
+
+//TODO: make this a macro
+int tqBufHandleSSize() {
+  return sizeof(int64_t) * 2
+    + sizeof(int32_t) * 2
+    + TQ_BUFFER_SIZE * tqBufItemSSize();
 }
-int tqBufItemSSize(const tqBufferItem *bufItem) {
+
+int tqBufItemSSize() {
+  //TODO: do this need serialization?
+  //mainly for executor
   return 0;
 }

source/server/vnode/tq/src/tqCommit.c

+/*
+ * 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/>.
+ */

source/server/vnode/tq/src/tqMetaStore.c

+/*
+ * 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 "tqMetaStore.h"
+//TODO:replace by a abstract file layer
+#include <fcntl.h>
+#include <unistd.h>
+
+typedef struct TqMetaPageBuf {
+  int16_t offset;
+  char buffer[TQ_PAGE_SIZE];
+} TqMetaPageBuf;
+
+TqMetaStore* tqStoreOpen(const char* path, void* serializer(void*),
+    void* deserializer(void*), void deleter(void*)) {
+  //concat data file name and index file name
+  int fileFd = open(path, O_WRONLY | O_CREAT | O_EXCL, 0755);
+  if(fileFd < 0) return NULL;
+  TqMetaStore* pMeta = malloc(sizeof(TqMetaStore)); 
+  if(pMeta == NULL) {
+    //close
+    return NULL;
+  }
+  memset(pMeta, 0, sizeof(TqMetaStore));
+  pMeta->fileFd = fileFd;
+  
+  int idxFd = open(path, O_WRONLY | O_CREAT | O_EXCL, 0755);
+  if(idxFd < 0) {
+    //close file
+    //free memory
+    return NULL;
+  }
+  pMeta->idxFd = idxFd;
+  pMeta->unpersistHead = malloc(sizeof(TqMetaList));
+  if(pMeta->unpersistHead == NULL) {
+    //close file
+    //free memory
+    return NULL;
+  }
+  pMeta->serializer = serializer;
+  pMeta->deserializer = deserializer;
+  pMeta->deleter = deleter;
+  return pMeta;
+}
+
+int32_t tqStoreClose(TqMetaStore* pMeta) {
+  //commit data and idx
+  //close file
+  //free memory
+  return 0;
+}
+
+int32_t tqStoreDelete(TqMetaStore* pMeta) {
+  //close file
+  //delete file
+  //free memory
+  return 0;
+}
+
+int32_t tqStorePersist(TqMetaStore* pMeta) {
+  TqMetaList *node = pMeta->unpersistHead;
+  while(node->unpersistNext) {
+    //serialize
+    //append data
+    //write offset and idx
+    //remove from unpersist list
+  }
+  return 0;
+}
+
+int32_t tqHandlePutInUse(TqMetaStore* pMeta, TqMetaHandle* handle) {
+  return 0;
+}
+
+TqMetaHandle* tqHandleGetInUse(TqMetaStore* pMeta, int64_t key) {
+  return NULL;
+}
+
+int32_t tqHandlePutInTxn(TqMetaStore* pMeta, TqMetaHandle* handle) {
+  return 0;
+}
+
+TqMetaHandle* tqHandleGetInTxn(TqMetaStore* pMeta, int64_t key) {
+  return NULL;
+}
+
+int32_t tqHandleCommit(TqMetaStore* pMeta, int64_t key) {
+  return 0;
+}
+
+int32_t tqHandleAbort(TqMetaStore* pMeta, int64_t key) {
+  return 0;
+}
+
+int32_t tqHandleDel(TqMetaStore* pMeta, int64_t key) {
+  return 0;
+}
+
+int32_t tqHandleClear(TqMetaStore* pMeta, int64_t key) {
+  return 0;
+}