merge from 3.0

Jenkinsfile2

@@ -35,6 +35,7 @@ def abort_previous(){
 def pre_test(){
     sh'hostname'
     sh '''
+    date
     sudo rmtaos || echo "taosd has not installed"
     '''
     sh '''
@@ -60,8 +61,9 @@ def pre_test(){
         sh '''
         cd ${WKC}
         git checkout 3.0
+        [ -d contrib/bdb ] && cd contrib/bdb && git clean -fxd && cd ../..
         '''
-      } 
+      }
       else{
         sh '''
         cd ${WKC}
@@ -98,7 +100,7 @@ pipeline {
   }
   stages {
       stage('pre_build'){
-          agent{label 'slave3_0'}
+          agent{label " slave3_0 || slave15 || slave16 || slave17 "}
           options { skipDefaultCheckout() } 
           when {
               changeRequest()

cmake/cmake.define

@@ -36,7 +36,14 @@ IF (TD_WINDOWS)
     ENDIF ()
 
 ELSE ()
-    SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -fPIC -gdwarf-2 -msse4.2 -mfma -g3")
-    SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -fPIC -gdwarf-2 -msse4.2 -mfma -g3")
+    SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -fPIC -gdwarf-2 -g3")
+    SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -fPIC -gdwarf-2 -g3")
+
+MESSAGE("System processor ID: ${CMAKE_SYSTEM_PROCESSOR}")
+IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64")
+    ADD_DEFINITIONS("-D_TD_ARM_")
+ELSE ()
+    ADD_DEFINITIONS("-msse4.2 -mfma")
+ENDIF ()
 
 ENDIF ()

cmake/cmake.platform

@@ -7,30 +7,52 @@ SET(TD_LINUX FALSE)
 SET(TD_WINDOWS FALSE)
 SET(TD_DARWIN FALSE)
 
-IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
+MESSAGE("Compiler ID: ${CMAKE_CXX_COMPILER_ID}")
+if(CMAKE_COMPILER_IS_GNUCXX MATCHES 1)
+    set(CXX_COMPILER_IS_GNU TRUE)
+else()
+    set(CXX_COMPILER_IS_GNU FALSE)
+endif()
 
-    SET(TD_LINUX TRUE)
-    SET(OSTYPE "Linux")
-    ADD_DEFINITIONS("-DLINUX")
+MESSAGE("Current system name is ${CMAKE_SYSTEM_NAME}.")
 
-    IF (${CMAKE_SIZEOF_VOID_P} MATCHES 8)
-        SET(TD_LINUX_64 TRUE)
+IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux" OR ${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
+
+    IF (${CXX_COMPILER_IS_GNU})
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -std=gnu99")
     ELSE ()
-        SET(TD_LINUX_32 TRUE)
+        ADD_DEFINITIONS("-Wno-tautological-constant-out-of-range-compare -Wno-pointer-sign -Wno-unknown-warning-option")
+        set(CMAKE_SHARED_LIBRARY_CREATE_C_FLAGS "${CMAKE_SHARED_LIBRARY_CREATE_C_FLAGS} -undefined dynamic_lookup")
+        set(CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS "${CMAKE_SHARED_LIBRARY_CREATE_CXX_FLAGS} -undefined dynamic_lookup")
     ENDIF ()
 
-ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
+    IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
 
-    SET(TD_DARWIN TRUE)
-    SET(OSTYPE "macOS")
-    ADD_DEFINITIONS("-DDARWIN")
-    IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64")
-        MESSAGE("Current system arch is arm64")
-        SET(TD_DARWIN_64 TRUE)
-        ADD_DEFINITIONS("-D_TD_DARWIN_64")
-    ENDIF ()
+        SET(TD_LINUX TRUE)
+        SET(OSTYPE "Linux")
+        ADD_DEFINITIONS("-DLINUX")
+
+        IF (${CMAKE_SIZEOF_VOID_P} MATCHES 8)
+            SET(TD_LINUX_64 TRUE)
+        ELSE ()
+            SET(TD_LINUX_32 TRUE)
+        ENDIF ()
+
+    ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
 
-    ADD_DEFINITIONS("-DHAVE_UNISTD_H")
+        SET(TD_DARWIN TRUE)
+        SET(OSTYPE "macOS")
+        ADD_DEFINITIONS("-DDARWIN -Wno-tautological-pointer-compare -Wno-return-type")
+
+        MESSAGE("Current system processor is ${CMAKE_SYSTEM_PROCESSOR}.")
+        IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64" OR ${CMAKE_SYSTEM_PROCESSOR} MATCHES "x86_64")
+            MESSAGE("Current system arch is arm64")
+            SET(TD_DARWIN_64 TRUE)
+            ADD_DEFINITIONS("-D_TD_DARWIN_64")
+        ENDIF ()
+
+        ADD_DEFINITIONS("-DHAVE_UNISTD_H")
+    ENDIF ()
 
 ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Windows")
 
@@ -45,6 +67,7 @@ ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Windows")
         SET(TD_WINDOWS_32 TRUE)
     	ADD_DEFINITIONS("-D_TD_WINDOWS_32")
     ENDIF ()
+
 ENDIF()
 
 MESSAGE("C Compiler ID: ${CMAKE_C_COMPILER_ID}")

contrib/CMakeLists.txt

@@ -127,6 +127,11 @@ target_include_directories(
 # zlib
 set(CMAKE_PROJECT_INCLUDE_BEFORE "${CMAKE_SUPPORT_DIR}/EnableCMP0048.txt.in")
 add_subdirectory(zlib)
+target_include_directories(
+    zlibstatic
+    PUBLIC ${CMAKE_CURRENT_BINARY_DIR}/zlib
+    PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/zlib
+)
 target_include_directories(
     zlib
     PUBLIC ${CMAKE_CURRENT_BINARY_DIR}/zlib

include/common/tcommon.h

@@ -54,14 +54,12 @@ typedef struct SColumnDataAgg {
 } SColumnDataAgg;
 
 typedef struct SDataBlockInfo {
-  STimeWindow window;
-  int32_t     rows;
-  int32_t     rowSize;
-  int32_t     numOfCols;
-  union {
-    int64_t uid;
-    int64_t blockId;
-  };
+  STimeWindow    window;
+  int32_t        rows;
+  int32_t        rowSize;
+  int16_t        numOfCols;
+  int16_t        hasVarCol;
+  union {int64_t uid; int64_t blockId;};
 } SDataBlockInfo;
 
 typedef struct SConstantItem {
@@ -73,10 +71,10 @@ typedef struct SConstantItem {
 
 // info.numOfCols = taosArrayGetSize(pDataBlock) + taosArrayGetSize(pConstantList);
 typedef struct SSDataBlock {
-  SColumnDataAgg* pBlockAgg;
-  SArray*         pDataBlock;  // SArray<SColumnInfoData>
-  SArray* pConstantList;       // SArray<SConstantItem>, it is a constant/tags value of the corresponding result value.
-  SDataBlockInfo info;
+  SColumnDataAgg *pBlockAgg;
+  SArray         *pDataBlock;    // SArray<SColumnInfoData>
+  SArray         *pConstantList; // SArray<SConstantItem>, it is a constant/tags value of the corresponding result value.
+  SDataBlockInfo  info;
 } SSDataBlock;
 
 typedef struct SVarColAttr {
@@ -99,13 +97,15 @@ typedef struct SColumnInfoData {
 
 static FORCE_INLINE int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
   int64_t tbUid = pBlock->info.uid;
-  int32_t numOfCols = pBlock->info.numOfCols;
+  int16_t numOfCols = pBlock->info.numOfCols;
+  int16_t hasVarCol = pBlock->info.hasVarCol;
   int32_t rows = pBlock->info.rows;
   int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
 
   int32_t tlen = 0;
   tlen += taosEncodeFixedI64(buf, tbUid);
-  tlen += taosEncodeFixedI32(buf, numOfCols);
+  tlen += taosEncodeFixedI16(buf, numOfCols);
+  tlen += taosEncodeFixedI16(buf, hasVarCol);
   tlen += taosEncodeFixedI32(buf, rows);
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
@@ -123,7 +123,8 @@ static FORCE_INLINE void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock)
   int32_t sz;
 
   buf = taosDecodeFixedI64(buf, &pBlock->info.uid);
-  buf = taosDecodeFixedI32(buf, &pBlock->info.numOfCols);
+  buf = taosDecodeFixedI16(buf, &pBlock->info.numOfCols);
+  buf = taosDecodeFixedI16(buf, &pBlock->info.hasVarCol);
   buf = taosDecodeFixedI32(buf, &pBlock->info.rows);
   buf = taosDecodeFixedI32(buf, &sz);
   pBlock->pDataBlock = taosArrayInit(sz, sizeof(SColumnInfoData));
@@ -247,7 +248,7 @@ typedef struct SGroupbyExpr {
 
 typedef struct SFunctParam {
   int32_t  type;
-  SColumn* pCol;
+  SColumn *pCol;
   SVariant param;
 } SFunctParam;
 
@@ -265,12 +266,12 @@ typedef struct SResSchame {
 typedef struct SExprBasicInfo {
   SResSchema   resSchema;
   int16_t      numOfParams;  // argument value of each function
-  SFunctParam* pParam;
+  SFunctParam *pParam;
 } SExprBasicInfo;
 
 typedef struct SExprInfo {
-  struct SExprBasicInfo base;
-  struct tExprNode*     pExpr;
+  struct SExprBasicInfo  base;
+  struct tExprNode      *pExpr;
 } SExprInfo;
 
 typedef struct SStateWindow {

include/common/tdatablock.h

@@ -117,7 +117,7 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
 
 int32_t      blockDataEnsureColumnCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
 int32_t      blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
-void         blockDataClearup(SSDataBlock* pDataBlock, bool hasVarCol);
+void         blockDataClearup(SSDataBlock* pDataBlock);
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock);
 size_t       blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
 void*        blockDataDestroy(SSDataBlock* pBlock);

include/common/tmsg.h

@@ -1151,8 +1151,8 @@ typedef struct {
   char   name[TSDB_TOPIC_FNAME_LEN];
   int8_t igExists;
   char*  sql;
-  char*  physicalPlan;
-  char*  logicalPlan;
+  char*  ast;
+  char   subscribeDbName[TSDB_DB_NAME_LEN];
 } SCMCreateTopicReq;
 
 int32_t tSerializeSCMCreateTopicReq(void* buf, int32_t bufLen, const SCMCreateTopicReq* pReq);
@@ -1911,11 +1911,12 @@ typedef struct {
   int8_t   slidingUnit;
   char     indexName[TSDB_INDEX_NAME_LEN];
   char     timezone[TD_TIMEZONE_LEN];  // sma data is invalid if timezone change.
-  uint16_t exprLen;
-  uint16_t tagsFilterLen;
+  int32_t  exprLen;
+  int32_t  tagsFilterLen;
   int64_t  indexUid;
   tb_uid_t tableUid;  // super/child/common table uid
   int64_t  interval;
+  int64_t  offset;
   int64_t  sliding;
   char*    expr;  // sma expression
   char*    tagsFilter;
@@ -2017,11 +2018,12 @@ static FORCE_INLINE int32_t tEncodeTSma(void** buf, const STSma* pSma) {
   tlen += taosEncodeFixedI8(buf, pSma->slidingUnit);
   tlen += taosEncodeString(buf, pSma->indexName);
   tlen += taosEncodeString(buf, pSma->timezone);
-  tlen += taosEncodeFixedU16(buf, pSma->exprLen);
-  tlen += taosEncodeFixedU16(buf, pSma->tagsFilterLen);
+  tlen += taosEncodeFixedI32(buf, pSma->exprLen);
+  tlen += taosEncodeFixedI32(buf, pSma->tagsFilterLen);
   tlen += taosEncodeFixedI64(buf, pSma->indexUid);
   tlen += taosEncodeFixedI64(buf, pSma->tableUid);
   tlen += taosEncodeFixedI64(buf, pSma->interval);
+  tlen += taosEncodeFixedI64(buf, pSma->offset);
   tlen += taosEncodeFixedI64(buf, pSma->sliding);
 
   if (pSma->exprLen > 0) {
@@ -2051,11 +2053,12 @@ static FORCE_INLINE void* tDecodeTSma(void* buf, STSma* pSma) {
   buf = taosDecodeFixedI8(buf, &pSma->slidingUnit);
   buf = taosDecodeStringTo(buf, pSma->indexName);
   buf = taosDecodeStringTo(buf, pSma->timezone);
-  buf = taosDecodeFixedU16(buf, &pSma->exprLen);
-  buf = taosDecodeFixedU16(buf, &pSma->tagsFilterLen);
+  buf = taosDecodeFixedI32(buf, &pSma->exprLen);
+  buf = taosDecodeFixedI32(buf, &pSma->tagsFilterLen);
   buf = taosDecodeFixedI64(buf, &pSma->indexUid);
   buf = taosDecodeFixedI64(buf, &pSma->tableUid);
   buf = taosDecodeFixedI64(buf, &pSma->interval);
+  buf = taosDecodeFixedI64(buf, &pSma->offset);
   buf = taosDecodeFixedI64(buf, &pSma->sliding);
 
   if (pSma->exprLen > 0) {

include/common/ttime.h

@@ -25,6 +25,17 @@ extern "C" {
 
 #define TIME_IS_VAR_DURATION(_t) ((_t) == 'n' || (_t) == 'y' || (_t) == 'N' || (_t) == 'Y')
 
+#define TIME_UNIT_NANOSECOND  'b'
+#define TIME_UNIT_MICROSECOND 'u'
+#define TIME_UNIT_MILLISECOND 'a'
+#define TIME_UNIT_SECOND      's'
+#define TIME_UNIT_MINUTE      'm'
+#define TIME_UNIT_HOUR        'h'
+#define TIME_UNIT_DAY         'd'
+#define TIME_UNIT_WEEK        'w'
+#define TIME_UNIT_MONTH       'n'
+#define TIME_UNIT_YEAR        'y'
+
 /*
  * @return timestamp decided by global conf variable, tsTimePrecision
  * if precision == TSDB_TIME_PRECISION_MICRO, it returns timestamp in microsecond.

include/common/ttokendef.h

@@ -48,108 +48,114 @@
 #define TK_DNODES                          30
 #define TK_NK_ID                           31
 #define TK_NK_IPTOKEN                      32
-#define TK_DATABASE                        33
-#define TK_DATABASES                       34
-#define TK_USE                             35
-#define TK_IF                              36
-#define TK_NOT                             37
-#define TK_EXISTS                          38
-#define TK_BLOCKS                          39
-#define TK_CACHE                           40
-#define TK_CACHELAST                       41
-#define TK_COMP                            42
-#define TK_DAYS                            43
-#define TK_FSYNC                           44
-#define TK_MAXROWS                         45
-#define TK_MINROWS                         46
-#define TK_KEEP                            47
-#define TK_PRECISION                       48
-#define TK_QUORUM                          49
-#define TK_REPLICA                         50
-#define TK_TTL                             51
-#define TK_WAL                             52
-#define TK_VGROUPS                         53
-#define TK_SINGLE_STABLE                   54
-#define TK_STREAM_MODE                     55
-#define TK_TABLE                           56
-#define TK_NK_LP                           57
-#define TK_NK_RP                           58
-#define TK_STABLE                          59
-#define TK_TABLES                          60
-#define TK_STABLES                         61
-#define TK_USING                           62
-#define TK_TAGS                            63
-#define TK_NK_DOT                          64
-#define TK_NK_COMMA                        65
-#define TK_COMMENT                         66
-#define TK_BOOL                            67
-#define TK_TINYINT                         68
-#define TK_SMALLINT                        69
-#define TK_INT                             70
-#define TK_INTEGER                         71
-#define TK_BIGINT                          72
-#define TK_FLOAT                           73
-#define TK_DOUBLE                          74
-#define TK_BINARY                          75
-#define TK_TIMESTAMP                       76
-#define TK_NCHAR                           77
-#define TK_UNSIGNED                        78
-#define TK_JSON                            79
-#define TK_VARCHAR                         80
-#define TK_MEDIUMBLOB                      81
-#define TK_BLOB                            82
-#define TK_VARBINARY                       83
-#define TK_DECIMAL                         84
-#define TK_SMA                             85
-#define TK_MNODES                          86
-#define TK_NK_FLOAT                        87
-#define TK_NK_BOOL                         88
-#define TK_NK_VARIABLE                     89
-#define TK_BETWEEN                         90
-#define TK_IS                              91
-#define TK_NULL                            92
-#define TK_NK_LT                           93
-#define TK_NK_GT                           94
-#define TK_NK_LE                           95
-#define TK_NK_GE                           96
-#define TK_NK_NE                           97
-#define TK_NK_EQ                           98
-#define TK_LIKE                            99
-#define TK_MATCH                          100
-#define TK_NMATCH                         101
-#define TK_IN                             102
-#define TK_FROM                           103
-#define TK_AS                             104
-#define TK_JOIN                           105
-#define TK_ON                             106
-#define TK_INNER                          107
-#define TK_SELECT                         108
-#define TK_DISTINCT                       109
-#define TK_WHERE                          110
-#define TK_PARTITION                      111
-#define TK_BY                             112
-#define TK_SESSION                        113
-#define TK_STATE_WINDOW                   114
-#define TK_INTERVAL                       115
-#define TK_SLIDING                        116
-#define TK_FILL                           117
-#define TK_VALUE                          118
-#define TK_NONE                           119
-#define TK_PREV                           120
-#define TK_LINEAR                         121
-#define TK_NEXT                           122
-#define TK_GROUP                          123
-#define TK_HAVING                         124
-#define TK_ORDER                          125
-#define TK_SLIMIT                         126
-#define TK_SOFFSET                        127
-#define TK_LIMIT                          128
-#define TK_OFFSET                         129
-#define TK_ASC                            130
-#define TK_DESC                           131
-#define TK_NULLS                          132
-#define TK_FIRST                          133
-#define TK_LAST                           134
+#define TK_QNODE                           33
+#define TK_ON                              34
+#define TK_QNODES                          35
+#define TK_DATABASE                        36
+#define TK_DATABASES                       37
+#define TK_USE                             38
+#define TK_IF                              39
+#define TK_NOT                             40
+#define TK_EXISTS                          41
+#define TK_BLOCKS                          42
+#define TK_CACHE                           43
+#define TK_CACHELAST                       44
+#define TK_COMP                            45
+#define TK_DAYS                            46
+#define TK_FSYNC                           47
+#define TK_MAXROWS                         48
+#define TK_MINROWS                         49
+#define TK_KEEP                            50
+#define TK_PRECISION                       51
+#define TK_QUORUM                          52
+#define TK_REPLICA                         53
+#define TK_TTL                             54
+#define TK_WAL                             55
+#define TK_VGROUPS                         56
+#define TK_SINGLE_STABLE                   57
+#define TK_STREAM_MODE                     58
+#define TK_TABLE                           59
+#define TK_NK_LP                           60
+#define TK_NK_RP                           61
+#define TK_STABLE                          62
+#define TK_TABLES                          63
+#define TK_STABLES                         64
+#define TK_USING                           65
+#define TK_TAGS                            66
+#define TK_NK_DOT                          67
+#define TK_NK_COMMA                        68
+#define TK_COMMENT                         69
+#define TK_BOOL                            70
+#define TK_TINYINT                         71
+#define TK_SMALLINT                        72
+#define TK_INT                             73
+#define TK_INTEGER                         74
+#define TK_BIGINT                          75
+#define TK_FLOAT                           76
+#define TK_DOUBLE                          77
+#define TK_BINARY                          78
+#define TK_TIMESTAMP                       79
+#define TK_NCHAR                           80
+#define TK_UNSIGNED                        81
+#define TK_JSON                            82
+#define TK_VARCHAR                         83
+#define TK_MEDIUMBLOB                      84
+#define TK_BLOB                            85
+#define TK_VARBINARY                       86
+#define TK_DECIMAL                         87
+#define TK_SMA                             88
+#define TK_INDEX                           89
+#define TK_FULLTEXT                        90
+#define TK_FUNCTION                        91
+#define TK_INTERVAL                        92
+#define TK_TOPIC                           93
+#define TK_AS                              94
+#define TK_MNODES                          95
+#define TK_NK_FLOAT                        96
+#define TK_NK_BOOL                         97
+#define TK_NK_VARIABLE                     98
+#define TK_BETWEEN                         99
+#define TK_IS                             100
+#define TK_NULL                           101
+#define TK_NK_LT                          102
+#define TK_NK_GT                          103
+#define TK_NK_LE                          104
+#define TK_NK_GE                          105
+#define TK_NK_NE                          106
+#define TK_NK_EQ                          107
+#define TK_LIKE                           108
+#define TK_MATCH                          109
+#define TK_NMATCH                         110
+#define TK_IN                             111
+#define TK_FROM                           112
+#define TK_JOIN                           113
+#define TK_INNER                          114
+#define TK_SELECT                         115
+#define TK_DISTINCT                       116
+#define TK_WHERE                          117
+#define TK_PARTITION                      118
+#define TK_BY                             119
+#define TK_SESSION                        120
+#define TK_STATE_WINDOW                   121
+#define TK_SLIDING                        122
+#define TK_FILL                           123
+#define TK_VALUE                          124
+#define TK_NONE                           125
+#define TK_PREV                           126
+#define TK_LINEAR                         127
+#define TK_NEXT                           128
+#define TK_GROUP                          129
+#define TK_HAVING                         130
+#define TK_ORDER                          131
+#define TK_SLIMIT                         132
+#define TK_SOFFSET                        133
+#define TK_LIMIT                          134
+#define TK_OFFSET                         135
+#define TK_ASC                            136
+#define TK_DESC                           137
+#define TK_NULLS                          138
+#define TK_FIRST                          139
+#define TK_LAST                           140
 
 #define TK_NK_SPACE                       300
 #define TK_NK_COMMENT                     301

include/libs/function/function.h

@@ -295,19 +295,8 @@ typedef struct SMultiFunctionsDesc {
 int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, SResultDataInfo* pInfo, int16_t extLength,
                           bool isSuperTable);
 
-/**
- * If the given name is a valid built-in sql function, the value of true will be returned.
- * @param name
- * @param len
- * @return
- */
-int32_t qIsBuiltinFunction(const char* name, int32_t len, bool* scalarFunction);
-
 bool qIsValidUdf(SArray* pUdfInfo, const char* name, int32_t len, int32_t* functionId);
 
-bool qIsAggregateFunction(const char* functionName);
-bool qIsSelectivityFunction(const char* functionName);
-
 tExprNode* exprTreeFromBinary(const void* data, size_t size);
 
 void extractFunctionDesc(SArray* pFunctionIdList, SMultiFunctionsDesc* pDesc);

include/libs/nodes/cmdnodes.h

@@ -23,6 +23,7 @@ extern "C" {
 #include "querynodes.h"
 
 typedef struct SDatabaseOptions {
+  ENodeType type;
   int32_t numOfBlocks;
   int32_t cacheBlockSize;
   int8_t cachelast;
@@ -46,7 +47,7 @@ typedef struct SCreateDatabaseStmt {
   ENodeType type;
   char dbName[TSDB_DB_NAME_LEN];
   bool ignoreExists;
-  SDatabaseOptions options;
+  SDatabaseOptions* pOptions;
 } SCreateDatabaseStmt;
 
 typedef struct SUseDatabaseStmt {
@@ -60,7 +61,14 @@ typedef struct SDropDatabaseStmt {
   bool ignoreNotExists;
 } SDropDatabaseStmt;
 
+typedef struct SAlterDatabaseStmt {
+  ENodeType type;
+  char dbName[TSDB_DB_NAME_LEN];
+  SDatabaseOptions* pOptions;
+} SAlterDatabaseStmt;
+
 typedef struct STableOptions {
+  ENodeType type;
   int32_t keep;
   int32_t ttl;
   char comments[TSDB_STB_COMMENT_LEN];
@@ -81,7 +89,7 @@ typedef struct SCreateTableStmt {
   bool ignoreExists;
   SNodeList* pCols;
   SNodeList* pTags;
-  STableOptions options;
+  STableOptions* pOptions;
 } SCreateTableStmt;
 
 typedef struct SCreateSubTableClause {
@@ -155,6 +163,58 @@ typedef struct SShowStmt {
   char dbName[TSDB_DB_NAME_LEN];
 } SShowStmt;
 
+typedef enum EIndexType {
+  INDEX_TYPE_SMA = 1,
+  INDEX_TYPE_FULLTEXT
+} EIndexType;
+
+typedef struct SIndexOptions {
+  ENodeType type;
+  SNodeList* pFuncs;
+  SNode* pInterval;
+  SNode* pOffset;
+  SNode* pSliding;
+} SIndexOptions;
+
+typedef struct SCreateIndexStmt {
+  ENodeType type;
+  EIndexType indexType;
+  char indexName[TSDB_INDEX_NAME_LEN];
+  char tableName[TSDB_TABLE_NAME_LEN];
+  SNodeList* pCols;
+  SIndexOptions* pOptions;
+} SCreateIndexStmt;
+
+typedef struct SDropIndexStmt {
+  ENodeType type;
+  char indexName[TSDB_INDEX_NAME_LEN];
+  char tableName[TSDB_TABLE_NAME_LEN];
+} SDropIndexStmt;
+
+typedef struct SCreateQnodeStmt {
+  ENodeType type;
+  int32_t dnodeId;
+} SCreateQnodeStmt;
+
+typedef struct SDropQnodeStmt {
+  ENodeType type;
+  int32_t dnodeId;
+} SDropQnodeStmt;
+
+typedef struct SCreateTopicStmt {
+  ENodeType type;
+  char topicName[TSDB_TABLE_NAME_LEN];
+  char subscribeDbName[TSDB_DB_NAME_LEN];
+  bool ignoreExists;
+  SNode* pQuery;
+} SCreateTopicStmt;
+
+typedef struct SDropTopicStmt {
+  ENodeType type;
+  char topicName[TSDB_TABLE_NAME_LEN];
+  bool ignoreNotExists;
+} SDropTopicStmt;
+
 #ifdef __cplusplus
 }
 #endif

include/libs/nodes/nodes.h

@@ -67,6 +67,9 @@ typedef enum ENodeType {
   QUERY_NODE_SLOT_DESC,
   QUERY_NODE_COLUMN_DEF,
   QUERY_NODE_DOWNSTREAM_SOURCE,
+  QUERY_NODE_DATABASE_OPTIONS,
+  QUERY_NODE_TABLE_OPTIONS,
+  QUERY_NODE_INDEX_OPTIONS,
 
   // Statement nodes are used in parser and planner module.
   QUERY_NODE_SET_OPERATOR,
@@ -74,6 +77,7 @@ typedef enum ENodeType {
   QUERY_NODE_VNODE_MODIF_STMT,
   QUERY_NODE_CREATE_DATABASE_STMT,
   QUERY_NODE_DROP_DATABASE_STMT,
+  QUERY_NODE_ALTER_DATABASE_STMT,
   QUERY_NODE_SHOW_DATABASES_STMT, // temp
   QUERY_NODE_CREATE_TABLE_STMT,
   QUERY_NODE_CREATE_SUBTABLE_CLAUSE,
@@ -93,6 +97,13 @@ typedef enum ENodeType {
   QUERY_NODE_SHOW_DNODES_STMT,
   QUERY_NODE_SHOW_VGROUPS_STMT,
   QUERY_NODE_SHOW_MNODES_STMT,
+  QUERY_NODE_SHOW_QNODES_STMT,
+  QUERY_NODE_CREATE_INDEX_STMT,
+  QUERY_NODE_DROP_INDEX_STMT,
+  QUERY_NODE_CREATE_QNODE_STMT,
+  QUERY_NODE_DROP_QNODE_STMT,
+  QUERY_NODE_CREATE_TOPIC_STMT,
+  QUERY_NODE_DROP_TOPIC_STMT,
 
   // logic plan node
   QUERY_NODE_LOGIC_PLAN_SCAN,
@@ -185,6 +196,9 @@ const char* nodesNodeName(ENodeType type);
 int32_t nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_t* pLen);
 int32_t nodesStringToNode(const char* pStr, SNode** pNode);
 
+int32_t nodesListToString(const SNodeList* pList, bool format, char** pStr, int32_t* pLen);
+int32_t nodesStringToList(const char* pStr, SNodeList** pList);
+
 #ifdef __cplusplus
 }
 #endif

include/libs/nodes/plannodes.h

@@ -93,6 +93,8 @@ typedef struct SWindowLogicNode {
   int64_t interval;
   int64_t offset;
   int64_t sliding;
+  int8_t  intervalUnit;
+  int8_t  slidingUnit;
   SFillNode* pFill;
 } SWindowLogicNode;
 
@@ -203,7 +205,7 @@ typedef struct SDownstreamSourceNode {
 
 typedef struct SExchangePhysiNode {
   SPhysiNode node;
-  int32_t srcGroupId;  // group id of datasource suplans
+  int32_t    srcGroupId;  // group id of datasource suplans
   SNodeList* pSrcEndPoints;  // element is SDownstreamSource, scheduler fill by calling qSetSuplanExecutionNode
 } SExchangePhysiNode;
 
@@ -211,9 +213,11 @@ typedef struct SIntervalPhysiNode {
   SPhysiNode node;
   SNodeList* pExprs;   // these are expression list of parameter expression of function
   SNodeList* pFuncs;
-  int64_t interval;
-  int64_t offset;
-  int64_t sliding;
+  int64_t    interval;
+  int64_t    offset;
+  int64_t    sliding;
+  int8_t     intervalUnit;
+  int8_t     slidingUnit;
   SFillNode* pFill;
 } SIntervalPhysiNode;
 

include/libs/nodes/querynodes.h

@@ -82,6 +82,7 @@ typedef struct SValueNode {
     double d;
     char* p;
   } datum;
+  char unit;
 } SValueNode;
 
 typedef struct SOperatorNode {

include/libs/parser/parser.h

@@ -51,6 +51,7 @@ typedef struct SQuery {
   SSchema* pResSchema;
   SCmdMsgInfo* pCmdMsg;
   int32_t msgType;
+  bool streamQuery;
 } SQuery;
 
 int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery);

include/libs/planner/planner.h

@@ -26,6 +26,7 @@ typedef struct SPlanContext {
   uint64_t queryId;
   int32_t acctId;
   SNode* pAstRoot;
+  bool streamQuery;
 } SPlanContext;
 
 // Create the physical plan for the query, according to the AST.

include/libs/transport/trpc.h

@@ -29,7 +29,6 @@ extern "C" {
 
 extern int tsRpcHeadSize;
 
-
 typedef struct SRpcConnInfo {
   uint32_t clientIp;
   uint16_t clientPort;
@@ -46,7 +45,6 @@ typedef struct SRpcMsg {
   void *  ahandle;  // app handle set by client
 } SRpcMsg;
 
-
 typedef struct SRpcInit {
   uint16_t localPort;     // local port
   char *   label;         // for debug purpose
@@ -71,9 +69,11 @@ typedef struct SRpcInit {
   // call back to keep conn or not
   bool (*pfp)(void *parent, tmsg_t msgType);
 
-  // to support Send messages multiple times on a link  
-  // 
-  void* (*mfp)(void *parent, tmsg_t msgType);
+  // to support Send messages multiple times on a link
+  void *(*mfp)(void *parent, tmsg_t msgType);
+
+  // call back  to handle except when query/fetch in progress
+  bool (*efp)(void *parent, tmsg_t msgType);
 
   void *parent;
 } SRpcInit;
@@ -94,7 +94,7 @@ int     rpcReportProgress(void *pConn, char *pCont, int contLen);
 void    rpcCancelRequest(int64_t rid);
 
 // just release client conn to rpc instance, no close sock
-void rpcReleaseHandle(void *handle);   
+void rpcReleaseHandle(void *handle, int8_t type);
 
 void rpcRefHandle(void *handle, int8_t type);
 void rpcUnrefHandle(void *handle, int8_t type);

include/os/os.h

@@ -22,6 +22,8 @@ extern "C" {
 
 #include <assert.h>
 #include <ctype.h>
+#include <pthread.h>
+#include <semaphore.h>
 
 #if !defined(WINDOWS)
 #include <unistd.h>
@@ -34,7 +36,12 @@ extern "C" {
 #include <sys/utsname.h>
 #include <sys/param.h>
 #include <sys/mman.h>
+
+#if defined(DARWIN)
+#else
 #include <sys/prctl.h>
+#include <argp.h>
+#endif
 
 #endif
 
@@ -54,9 +61,12 @@ extern "C" {
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/ioctl.h>
 #include <sys/stat.h>
+#include <sys/time.h>
 #include <sys/types.h>
 #include <wchar.h>
+#include <termios.h>
 #include <wctype.h>
 
 

include/os/osEok.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_OS_EOK_H
+#define TDENGINE_OS_EOK_H
+
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#ifdef __APPLE__
+
+enum EPOLL_EVENTS
+  {
+    EPOLLIN = 0x001,
+#define EPOLLIN EPOLLIN
+    EPOLLPRI = 0x002,
+#define EPOLLPRI EPOLLPRI
+    EPOLLOUT = 0x004,
+#define EPOLLOUT EPOLLOUT
+    EPOLLRDNORM = 0x040,
+#define EPOLLRDNORM EPOLLRDNORM
+    EPOLLRDBAND = 0x080,
+#define EPOLLRDBAND EPOLLRDBAND
+    EPOLLWRNORM = 0x100,
+#define EPOLLWRNORM EPOLLWRNORM
+    EPOLLWRBAND = 0x200,
+#define EPOLLWRBAND EPOLLWRBAND
+    EPOLLMSG = 0x400,
+#define EPOLLMSG EPOLLMSG
+    EPOLLERR = 0x008,
+#define EPOLLERR EPOLLERR
+    EPOLLHUP = 0x010,
+#define EPOLLHUP EPOLLHUP
+    EPOLLRDHUP = 0x2000,
+#define EPOLLRDHUP EPOLLRDHUP
+    EPOLLEXCLUSIVE = 1u << 28,
+#define EPOLLEXCLUSIVE EPOLLEXCLUSIVE
+    EPOLLWAKEUP = 1u << 29,
+#define EPOLLWAKEUP EPOLLWAKEUP
+    EPOLLONESHOT = 1u << 30,
+#define EPOLLONESHOT EPOLLONESHOT
+    EPOLLET = 1u << 31
+#define EPOLLET EPOLLET
+  };
+
+/* Valid opcodes ( "op" parameter ) to issue to epoll_ctl().  */
+#define EPOLL_CTL_ADD 1 /* Add a file descriptor to the interface.  */
+#define EPOLL_CTL_DEL 2 /* Remove a file descriptor from the interface.  */
+#define EPOLL_CTL_MOD 3 /* Change file descriptor epoll_event structure.  */
+
+
+typedef union epoll_data
+{
+  void *ptr;
+  int fd;
+  uint32_t u32;
+  uint64_t u64;
+} epoll_data_t;
+
+struct epoll_event
+{
+  uint32_t events;  /* Epoll events */
+  epoll_data_t data;  /* User data variable */
+};
+
+int epoll_create(int size);
+int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
+int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
+int epoll_close(int epfd);
+
+#endif // __APPLE__
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // _eok_h_fd274616_996c_400e_9023_ae70be881fa3_
+

include/os/osFile.h

@@ -49,6 +49,15 @@ int64_t taosRead(FileFd fd, void *buf, int64_t count);
 #define PATH_MAX 256
 #endif
 
+typedef int32_t FileFd;
+
+typedef struct TdFile {
+  pthread_rwlock_t rwlock;
+  int      refId;
+  FileFd   fd;
+  FILE    *fp;
+} * TdFilePtr, TdFile;
+
 typedef struct TdFile *TdFilePtr;
  
 #define TD_FILE_CTEATE    0x0001
@@ -101,8 +110,16 @@ int64_t taosCopyFile(const char *from, const char *to);
 int32_t taosRemoveFile(const char *path);
  
 void    taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, char *dstPath);
- 
+
+#if defined(_TD_DARWIN_64)
+typedef int32_t SocketFd;
+
+int64_t taosSendFile(SocketFd fdDst, FileFd pFileSrc, int64_t *offset, int64_t size);
+int64_t taosFSendFile(FILE *pFileOut, FILE *pFileIn, int64_t *offset, int64_t size);
+#else
+int64_t taosSendFile(SocketFd fdDst, TdFilePtr pFileSrc, int64_t *offset, int64_t size);
 int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size);
+#endif
 
 void *taosMmapReadOnlyFile(TdFilePtr pFile, int64_t length);
 bool taosValidFile(TdFilePtr pFile);

include/os/osSemaphore.h

@@ -38,12 +38,12 @@ extern "C" {
 #endif
 
 #if defined (_TD_DARWIN_64)
-  #define pthread_rwlock_t pthread_mutex_t
-  #define pthread_rwlock_init(lock, NULL) pthread_mutex_init(lock, NULL)
-  #define pthread_rwlock_destroy(lock) pthread_mutex_destroy(lock)
-  #define pthread_rwlock_wrlock(lock) pthread_mutex_lock(lock)
-  #define pthread_rwlock_rdlock(lock) pthread_mutex_lock(lock)
-  #define pthread_rwlock_unlock(lock) pthread_mutex_unlock(lock)
+//  #define pthread_rwlock_t pthread_mutex_t
+//  #define pthread_rwlock_init(lock, NULL) pthread_mutex_init(lock, NULL)
+//  #define pthread_rwlock_destroy(lock) pthread_mutex_destroy(lock)
+// #define pthread_rwlock_wrlock(lock) pthread_mutex_lock(lock)
+//  #define pthread_rwlock_rdlock(lock) pthread_mutex_lock(lock)
+//  #define pthread_rwlock_unlock(lock) pthread_mutex_unlock(lock)
 
   #define pthread_spinlock_t pthread_mutex_t
   #define pthread_spin_init(lock, NULL) pthread_mutex_init(lock, NULL)

include/os/osSocket.h

@@ -33,8 +33,15 @@
   #include <winbase.h>
   #include <Winsock2.h>
 #else
-  #include <netinet/in.h>
-  #include <sys/epoll.h>
+    #include <netinet/in.h>
+    #include <sys/socket.h>
+
+    #if defined(_TD_DARWIN_64)
+        #include <osEok.h>
+    #else
+        #include <netinet/in.h>
+        #include <sys/epoll.h>
+    #endif
 #endif
 
 #ifdef __cplusplus
@@ -49,7 +56,29 @@ extern "C" {
 #endif
 
 #if defined(_TD_DARWIN_64)
-  #define htobe64 htonll
+//  #define htobe64 htonll
+
+#	include <libkern/OSByteOrder.h>
+
+#	define htobe16(x) OSSwapHostToBigInt16(x)
+#	define htole16(x) OSSwapHostToLittleInt16(x)
+#	define be16toh(x) OSSwapBigToHostInt16(x)
+#	define le16toh(x) OSSwapLittleToHostInt16(x)
+ 
+#	define htobe32(x) OSSwapHostToBigInt32(x)
+#	define htole32(x) OSSwapHostToLittleInt32(x)
+#	define be32toh(x) OSSwapBigToHostInt32(x)
+#	define le32toh(x) OSSwapLittleToHostInt32(x)
+ 
+#	define htobe64(x) OSSwapHostToBigInt64(x)
+#	define htole64(x) OSSwapHostToLittleInt64(x)
+#	define be64toh(x) OSSwapBigToHostInt64(x)
+#	define le64toh(x) OSSwapLittleToHostInt64(x)
+
+#	define __BYTE_ORDER    BYTE_ORDER
+#	define __BIG_ENDIAN    BIG_ENDIAN
+#	define __LITTLE_ENDIAN LITTLE_ENDIAN
+#	define __PDP_ENDIAN    PDP_ENDIAN
 #endif
 
 #define TAOS_EPOLL_WAIT_TIME 500

include/os/osSysinfo.h

@@ -16,6 +16,7 @@
 #ifndef _TD_OS_SYSINFO_H_
 #define _TD_OS_SYSINFO_H_
 
+#include <sys/statvfs.h>
 #include "os.h"
 
 #ifdef __cplusplus
@@ -52,7 +53,7 @@ int32_t taosGetSystemUUID(char *uid, int32_t uidlen);
 char   *taosGetCmdlineByPID(int32_t pid);
 void    taosSetCoreDump(bool enable);
 
-#if defined(WINDOWS)
+#if !defined(LINUX)
 
 #define _UTSNAME_LENGTH 65
 #define _UTSNAME_MACHINE_LENGTH _UTSNAME_LENGTH

include/util/tjson.h

@@ -25,6 +25,7 @@ extern "C" {
 typedef void SJson;
 
 SJson* tjsonCreateObject();
+SJson* tjsonCreateArray();
 void   tjsonDelete(SJson* pJson);
 
 SJson*  tjsonAddArrayToObject(SJson* pJson, const char* pName);

source/client/inc/clientInt.h

@@ -21,6 +21,7 @@ extern "C" {
 #endif
 
 #include "parser.h"
+#include "planner.h"
 #include "query.h"
 #include "taos.h"
 #include "tcommon.h"
@@ -229,6 +230,7 @@ void setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t
 int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest);
 
 int32_t parseSql(SRequestObj* pRequest, SQuery** pQuery);
+int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);
 
 // --- heartbeat
 // global, called by mgmt

source/client/src/clientImpl.c

 
 #include "clientInt.h"
 #include "clientLog.h"
-#include "parser.h"
-#include "planner.h"
 #include "scheduler.h"
 #include "tdatablock.h"
 #include "tdef.h"
@@ -211,7 +209,6 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
   }
 }
 
-
 int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList) {
   void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
 

source/client/src/clientMsgHandler.c

@@ -33,7 +33,7 @@ int32_t genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
   setErrno(pRequest, code);
 
   free(pMsg->pData);
-  sem_post(&pRequest->body.rspSem);
+  tsem_post(&pRequest->body.rspSem);
   return code;
 }
 
@@ -42,7 +42,7 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
   if (code != TSDB_CODE_SUCCESS) {
     free(pMsg->pData);
     setErrno(pRequest, code);
-    sem_post(&pRequest->body.rspSem);
+    tsem_post(&pRequest->body.rspSem);
     return code;
   }
 
@@ -78,7 +78,7 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
            pTscObj->pAppInfo->numOfConns);
 
   free(pMsg->pData);
-  sem_post(&pRequest->body.rspSem);
+  tsem_post(&pRequest->body.rspSem);
   return 0;
 }
 

source/client/src/tmq.c

@@ -486,17 +486,15 @@ TAOS_RES* tmq_create_topic(TAOS* taos, const char* topicName, const char* sql, i
   CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
   CHECK_CODE_GOTO(parseSql(pRequest, &pQueryNode), _return);
 
-  SQueryStmtInfo* pQueryStmtInfo = (SQueryStmtInfo*)pQueryNode;
-  pQueryStmtInfo->info.continueQuery = true;
+  pQueryNode->streamQuery = true;
 
   // todo check for invalid sql statement and return with error code
 
   SSchema* schema = NULL;
   int32_t  numOfCols = 0;
-  CHECK_CODE_GOTO(qCreateQueryDag(pQueryNode, &pRequest->body.pDag, &schema, &numOfCols, NULL, pRequest->requestId),
-                  _return);
+  CHECK_CODE_GOTO(getPlan(pRequest, pQueryNode, &pRequest->body.pDag, NULL), _return);
 
-  pStr = qDagToString(pRequest->body.pDag);
+  pStr = qQueryPlanToString(pRequest->body.pDag);
   if (pStr == NULL) {
     goto _return;
   }
@@ -506,7 +504,7 @@ TAOS_RES* tmq_create_topic(TAOS* taos, const char* topicName, const char* sql, i
   // The topic should be related to a database that the queried table is belonged to.
   SName name = {0};
   char  dbName[TSDB_DB_FNAME_LEN] = {0};
-  tNameGetFullDbName(&((SQueryStmtInfo*)pQueryNode)->pTableMetaInfo[0]->name, dbName);
+  // tNameGetFullDbName(&((SQueryStmtInfo*)pQueryNode)->pTableMetaInfo[0]->name, dbName);
 
   tNameFromString(&name, dbName, T_NAME_ACCT | T_NAME_DB);
   tNameFromString(&name, topicName, T_NAME_TABLE);

source/common/src/tdatablock.c

@@ -864,7 +864,7 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirs
         qsort(pColInfoData->pData, pDataBlock->info.rows, pColInfoData->info.bytes, fn);
 
         int64_t p1 = taosGetTimestampUs();
-        printf("sort:%ld, rows:%d\n", p1 - p0, pDataBlock->info.rows);
+        printf("sort:%" PRId64 ", rows:%d\n", p1 - p0, pDataBlock->info.rows);
 
         return TSDB_CODE_SUCCESS;
       } else {  // var data type
@@ -912,7 +912,7 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirs
   copyBackToBlock(pDataBlock, pCols);
   int64_t p4 = taosGetTimestampUs();
 
-  printf("sort:%ld, create:%ld, assign:%ld, copyback:%ld, rows:%d\n", p1-p0, p2 - p1, p3 - p2, p4-p3, rows);
+  printf("sort:%" PRId64 ", create:%" PRId64 ", assign:%" PRId64 ", copyback:%" PRId64 ", rows:%d\n", p1-p0, p2 - p1, p3 - p2, p4-p3, rows);
   destroyTupleIndex(index);
 
   return TSDB_CODE_SUCCESS;
@@ -1017,7 +1017,7 @@ int32_t dataBlockCompar_rv(const void* p1, const void* p2, const void* param) {
 }
 
 int32_t varColSort(SColumnInfoData* pColumnInfoData, SBlockOrderInfo* pOrder) {
-
+    return 0;
 }
 
 int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst) {
@@ -1055,14 +1055,15 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
   copyBackToBlock(pDataBlock, pCols);
   int64_t p4 = taosGetTimestampUs();
 
-  printf("sort:%ld, create:%ld, assign:%ld, copyback:%ld, rows:%d\n", p1 - p0, p2 - p1, p3 - p2, p4 - p3, rows);
+  printf("sort:%" PRId64 ", create:%" PRId64", assign:%" PRId64 ", copyback:%" PRId64 ", rows:%d\n", p1 - p0, p2 - p1, p3 - p2, p4 - p3, rows);
   //  destroyTupleIndex(index);
+  return 0;
 }
 
-void blockDataClearup(SSDataBlock* pDataBlock, bool hasVarCol) {
+void blockDataClearup(SSDataBlock* pDataBlock) {
   pDataBlock->info.rows = 0;
 
-  if (hasVarCol) {
+  if (pDataBlock->info.hasVarCol) {
     for (int32_t i = 0; i < pDataBlock->info.numOfCols; ++i) {
       SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
 
@@ -1148,7 +1149,9 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock) {
 
   SSDataBlock* pBlock = calloc(1, sizeof(SSDataBlock));
   pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
+
   pBlock->info.numOfCols = numOfCols;
+  pBlock->info.hasVarCol = pDataBlock->info.hasVarCol;
 
   for(int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData  colInfo = {0};

source/common/src/tmsg.c

@@ -1989,11 +1989,9 @@ int32_t tDeserializeSMDropTopicReq(void *buf, int32_t bufLen, SMDropTopicReq *pR
 
 int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTopicReq *pReq) {
   int32_t sqlLen = 0;
-  int32_t physicalPlanLen = 0;
-  int32_t logicalPlanLen = 0;
+  int32_t astLen = 0;
   if (pReq->sql != NULL) sqlLen = (int32_t)strlen(pReq->sql);
-  if (pReq->physicalPlan != NULL) physicalPlanLen = (int32_t)strlen(pReq->physicalPlan);
-  if (pReq->logicalPlan != NULL) logicalPlanLen = (int32_t)strlen(pReq->logicalPlan);
+  if (pReq->ast != NULL) astLen = (int32_t)strlen(pReq->ast);
 
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
@@ -2002,11 +2000,9 @@ int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTo
   if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->igExists) < 0) return -1;
   if (tEncodeI32(&encoder, sqlLen) < 0) return -1;
-  if (tEncodeI32(&encoder, physicalPlanLen) < 0) return -1;
-  if (tEncodeI32(&encoder, logicalPlanLen) < 0) return -1;
-  if (tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
-  if (tEncodeCStr(&encoder, pReq->physicalPlan) < 0) return -1;
-  if (tEncodeCStr(&encoder, pReq->logicalPlan) < 0) return -1;
+  if (tEncodeI32(&encoder, astLen) < 0) return -1;
+  if (sqlLen > 0 && tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
+  if (astLen > 0 && tEncodeCStr(&encoder, pReq->ast) < 0) return -1;
 
   tEndEncode(&encoder);
 
@@ -2017,8 +2013,7 @@ int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTo
 
 int32_t tDeserializeSCMCreateTopicReq(void *buf, int32_t bufLen, SCMCreateTopicReq *pReq) {
   int32_t sqlLen = 0;
-  int32_t physicalPlanLen = 0;
-  int32_t logicalPlanLen = 0;
+  int32_t astLen = 0;
 
   SCoder decoder = {0};
   tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_DECODER);
@@ -2027,17 +2022,20 @@ int32_t tDeserializeSCMCreateTopicReq(void *buf, int32_t bufLen, SCMCreateTopicR
   if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->igExists) < 0) return -1;
   if (tDecodeI32(&decoder, &sqlLen) < 0) return -1;
-  if (tDecodeI32(&decoder, &physicalPlanLen) < 0) return -1;
-  if (tDecodeI32(&decoder, &logicalPlanLen) < 0) return -1;
+  if (tDecodeI32(&decoder, &astLen) < 0) return -1;
+
+  if (sqlLen > 0) {
+    pReq->sql = calloc(1, sqlLen + 1);
+    if (pReq->sql == NULL) return -1;
+    if (tDecodeCStrTo(&decoder, pReq->sql) < 0) return -1;
+  }
 
-  pReq->sql = calloc(1, sqlLen + 1);
-  pReq->physicalPlan = calloc(1, physicalPlanLen + 1);
-  pReq->logicalPlan = calloc(1, logicalPlanLen + 1);
-  if (pReq->sql == NULL || pReq->physicalPlan == NULL || pReq->logicalPlan == NULL) return -1;
+  if (astLen > 0) {
+    pReq->ast = calloc(1, astLen + 1);
+    if (pReq->ast == NULL) return -1;
+    if (tDecodeCStrTo(&decoder, pReq->ast) < 0) return -1;
+  }
 
-  if (tDecodeCStrTo(&decoder, pReq->sql) < 0) return -1;
-  if (tDecodeCStrTo(&decoder, pReq->physicalPlan) < 0) return -1;
-  if (tDecodeCStrTo(&decoder, pReq->logicalPlan) < 0) return -1;
   tEndDecode(&decoder);
 
   tCoderClear(&decoder);
@@ -2046,8 +2044,7 @@ int32_t tDeserializeSCMCreateTopicReq(void *buf, int32_t bufLen, SCMCreateTopicR
 
 void tFreeSCMCreateTopicReq(SCMCreateTopicReq *pReq) {
   tfree(pReq->sql);
-  tfree(pReq->physicalPlan);
-  tfree(pReq->logicalPlan);
+  tfree(pReq->ast);
 }
 
 int32_t tSerializeSCMCreateTopicRsp(void *buf, int32_t bufLen, const SCMCreateTopicRsp *pRsp) {

source/common/src/trow.c

@@ -353,10 +353,10 @@ static void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, i
         for (int i = 0; i < src2->numOfCols; i++) {
           SCellVal sVal = {0};
           ASSERT(target->cols[i].type == src2->cols[i].type);
-          if (src2->cols[i].len > 0 && !isNull(src2->cols[i].pData, src2->cols[i].type)) {
-            if (tdGetColDataOfRow(&sVal, src1->cols + i, *iter1) < 0) {
-              TASSERT(0);
-            }
+          if (tdGetColDataOfRow(&sVal, src2->cols + i, *iter2) < 0) {
+            TASSERT(0);
+          }
+          if (src2->cols[i].len > 0 && !tdValTypeIsNull(sVal.valType)) {
             tdAppendValToDataCol(&(target->cols[i]), sVal.valType, sVal.val, target->numOfRows, target->maxPoints);
           } else if (!forceSetNull && key1 == key2 && src1->cols[i].len > 0) {
             if (tdGetColDataOfRow(&sVal, src1->cols + i, *iter1) < 0) {

source/common/src/ttszip.c

@@ -845,7 +845,11 @@ int32_t tsBufMerge(STSBuf* pDestBuf, const STSBuf* pSrcBuf) {
 
   int64_t offset = getDataStartOffset();
   int32_t size = (int32_t)pSrcBuf->fileSize - (int32_t)offset;
+#if defined(_TD_DARWIN_64)
+  int64_t written = taosFSendFile(pDestBuf->pFile->fp, pSrcBuf->pFile->fp, &offset, size);
+#else
   int64_t written = taosFSendFile(pDestBuf->pFile, pSrcBuf->pFile, &offset, size);
+#endif
 
   if (written == -1 || written != size) {
     return -1;

source/dnode/mnode/impl/src/mndConsumer.c

@@ -96,12 +96,12 @@ SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {
 CM_ENCODE_OVER:
   tfree(buf);
   if (terrno != 0) {
-    mError("consumer:%ld, failed to encode to raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());
+    mError("consumer:%" PRId64 ", failed to encode to raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());
     sdbFreeRaw(pRaw);
     return NULL;
   }
 
-  mTrace("consumer:%ld, encode to raw:%p, row:%p", pConsumer->consumerId, pRaw, pConsumer);
+  mTrace("consumer:%" PRId64 ", encode to raw:%p, row:%p", pConsumer->consumerId, pRaw, pConsumer);
   return pRaw;
 }
 
@@ -140,7 +140,7 @@ SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {
 CM_DECODE_OVER:
   tfree(buf);
   if (terrno != TSDB_CODE_SUCCESS) {
-    mError("consumer:%ld, failed to decode from raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());
+    mError("consumer:%" PRId64 ", failed to decode from raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());
     tfree(pRow);
     return NULL;
   }
@@ -149,17 +149,17 @@ CM_DECODE_OVER:
 }
 
 static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {
-  mTrace("consumer:%ld, perform insert action", pConsumer->consumerId);
+  mTrace("consumer:%" PRId64 ", perform insert action", pConsumer->consumerId);
   return 0;
 }
 
 static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer) {
-  mTrace("consumer:%ld, perform delete action", pConsumer->consumerId);
+  mTrace("consumer:%" PRId64 ", perform delete action", pConsumer->consumerId);
   return 0;
 }
 
 static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer) {
-  mTrace("consumer:%ld, perform update action", pOldConsumer->consumerId);
+  mTrace("consumer:%" PRId64 ", perform update action", pOldConsumer->consumerId);
 
   // TODO handle update
   /*taosWLockLatch(&pOldConsumer->lock);*/

source/dnode/mnode/impl/src/mndDb.c

@@ -1292,11 +1292,11 @@ static int32_t mndGetDbMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMe
   pSchema[cols].bytes = pShow->bytes[cols];
   cols++;
 
-  pShow->bytes[cols] = 1;
-  pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
-  strcpy(pSchema[cols].name, "update");
-  pSchema[cols].bytes = pShow->bytes[cols];
-  cols++;
+//  pShow->bytes[cols] = 1;
+//  pSchema[cols].type = TSDB_DATA_TYPE_TINYINT;
+//  strcpy(pSchema[cols].name, "update");
+//  pSchema[cols].bytes = pShow->bytes[cols];
+//  cols++;
 
   pMeta->numOfColumns = cols;
   pShow->numOfColumns = cols;
@@ -1432,9 +1432,9 @@ static int32_t mndRetrieveDbs(SMnodeMsg *pReq, SShowObj *pShow, char *data, int3
     STR_WITH_SIZE_TO_VARSTR(pWrite, prec, 2);
     cols++;
 
-    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    *(int8_t *)pWrite = pDb->cfg.update;
-    cols++;
+//    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+//    *(int8_t *)pWrite = pDb->cfg.update;
+//    cols++;
 
     numOfRows++;
     sdbRelease(pSdb, pDb);

source/dnode/mnode/impl/src/mndSubscribe.c

@@ -448,7 +448,7 @@ static int32_t mndProcessDoRebalanceMsg(SMnodeMsg *pMsg) {
     for (int32_t i = 0; i < taosArrayGetSize(pRebSub->lostConsumers); i++) {
       int64_t lostConsumerId = *(int64_t *)taosArrayGet(pRebSub->lostConsumers, i);
 
-      mInfo("mq remove lost consumer %ld", lostConsumerId);
+      mInfo("mq remove lost consumer %" PRId64 "", lostConsumerId);
 
       for (int32_t j = 0; j < taosArrayGetSize(pSub->consumers); j++) {
         SMqSubConsumer *pSubConsumer = taosArrayGet(pSub->consumers, j);
@@ -479,7 +479,7 @@ static int32_t mndProcessDoRebalanceMsg(SMnodeMsg *pMsg) {
         else
           vgThisConsumerAfterRb = vgEachConsumer;
 
-        mInfo("mq consumer:%ld, connectted vgroup number change from %d to %d", pSubConsumer->consumerId,
+        mInfo("mq consumer:%" PRId64 ", connectted vgroup number change from %d to %d", pSubConsumer->consumerId,
               vgThisConsumerBeforeRb, vgThisConsumerAfterRb);
 
         while (taosArrayGetSize(pSubConsumer->vgInfo) > vgThisConsumerAfterRb) {
@@ -503,7 +503,7 @@ static int32_t mndProcessDoRebalanceMsg(SMnodeMsg *pMsg) {
             atomic_store_32(&pRebConsumer->status, MQ_CONSUMER_STATUS__IDLE);
           }
 
-          mInfo("mq consumer:%ld, status change from %d to %d", pRebConsumer->consumerId, status, pRebConsumer->status);
+          mInfo("mq consumer:%" PRId64 ", status change from %d to %d", pRebConsumer->consumerId, status, pRebConsumer->status);
 
           SSdbRaw *pConsumerRaw = mndConsumerActionEncode(pRebConsumer);
           sdbSetRawStatus(pConsumerRaw, SDB_STATUS_READY);
@@ -537,13 +537,13 @@ static int32_t mndProcessDoRebalanceMsg(SMnodeMsg *pMsg) {
               mndSplitSubscribeKey(pSub->key, topic, cgroup);
               SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
 
-              mInfo("mq set conn: assign vgroup %d of topic %s to consumer %ld", pConsumerEp->vgId, topic,
+              mInfo("mq set conn: assign vgroup %d of topic %s to consumer %" PRId64 "", pConsumerEp->vgId, topic,
                     pConsumerEp->consumerId);
 
               mndPersistMqSetConnReq(pMnode, pTrans, pTopic, cgroup, pConsumerEp);
               mndReleaseTopic(pMnode, pTopic);
             } else {
-              mInfo("mq rebalance: assign vgroup %d, from consumer %ld to consumer %ld", pConsumerEp->vgId,
+              mInfo("mq rebalance: assign vgroup %d, from consumer %" PRId64 " to consumer %" PRId64 "", pConsumerEp->vgId,
                     pConsumerEp->oldConsumerId, pConsumerEp->consumerId);
 
               mndPersistRebalanceMsg(pMnode, pTrans, pConsumerEp);
@@ -1099,7 +1099,7 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
       SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, cgroup, newTopicName);
       bool             createSub = false;
       if (pSub == NULL) {
-        mDebug("create new subscription by consumer %ld, group: %s, topic %s", consumerId, cgroup, newTopicName);
+        mDebug("create new subscription by consumer %" PRId64 ", group: %s, topic %s", consumerId, cgroup, newTopicName);
         pSub = mndCreateSubscription(pMnode, pTopic, cgroup);
         createSub = true;
 
@@ -1118,7 +1118,7 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
         pConsumerEp->consumerId = consumerId;
         taosArrayPush(mqSubConsumer.vgInfo, pConsumerEp);
         if (pConsumerEp->oldConsumerId == -1) {
-          mInfo("mq set conn: assign vgroup %d of topic %s to consumer %ld", pConsumerEp->vgId, newTopicName,
+          mInfo("mq set conn: assign vgroup %d of topic %s to consumer %" PRId64 "", pConsumerEp->vgId, newTopicName,
                 pConsumerEp->consumerId);
           mndPersistMqSetConnReq(pMnode, pTrans, pTopic, cgroup, pConsumerEp);
         } else {

source/dnode/mnode/impl/src/mndTopic.c

@@ -236,6 +236,29 @@ static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
   return 0;
 }
 
+static int32_t mndGetPlanString(SCMCreateTopicReq *pCreate, char **pStr) {
+  if (NULL == pCreate->ast) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SNode* pAst = NULL;
+  int32_t code = nodesStringToNode(pCreate->ast, &pAst);
+
+  SQueryPlan* pPlan = NULL;
+  if (TSDB_CODE_SUCCESS == code) {
+    SPlanContext cxt = { .pAstRoot = pAst, .streamQuery = true };
+    code = qCreateQueryPlan(&cxt, &pPlan, NULL);
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodesNodeToString(pPlan, false, pStr, NULL);
+  }
+  nodesDestroyNode(pAst);
+  nodesDestroyNode(pPlan);
+  terrno = code;
+  return code;
+}
+
 static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateTopicReq *pCreate, SDbObj *pDb) {
   mDebug("topic:%s to create", pCreate->name);
   SMqTopicObj topicObj = {0};
@@ -247,13 +270,23 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateTopicReq
   topicObj.dbUid = pDb->uid;
   topicObj.version = 1;
   topicObj.sql = pCreate->sql;
-  topicObj.physicalPlan = pCreate->physicalPlan;
-  topicObj.logicalPlan = pCreate->logicalPlan;
+  topicObj.physicalPlan = "";
+  topicObj.logicalPlan = "";
   topicObj.sqlLen = strlen(pCreate->sql);
 
+  char* pPlanStr = NULL;
+  if (TSDB_CODE_SUCCESS != mndGetPlanString(pCreate, &pPlanStr)) {
+    mError("topic:%s, failed to get plan since %s", pCreate->name, terrstr());
+    return -1;
+  }
+  if (NULL != pPlanStr) {
+    topicObj.physicalPlan = pPlanStr;
+  }
+
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_TOPIC, &pReq->rpcMsg);
   if (pTrans == NULL) {
     mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+    tfree(pPlanStr);
     return -1;
   }
   mDebug("trans:%d, used to create topic:%s", pTrans->id, pCreate->name);
@@ -261,6 +294,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateTopicReq
   SSdbRaw *pRedoRaw = mndTopicActionEncode(&topicObj);
   if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
     mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+    tfree(pPlanStr);
     mndTransDrop(pTrans);
     return -1;
   }
@@ -268,10 +302,12 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateTopicReq
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
+    tfree(pPlanStr);
     mndTransDrop(pTrans);
     return -1;
   }
 
+  tfree(pPlanStr);
   mndTransDrop(pTrans);
   return 0;
 }

source/dnode/mnode/impl/test/db/db.cpp

@@ -27,7 +27,7 @@ Testbase MndTestDb::test;
 
 TEST_F(MndTestDb, 01_ShowDb) {
   test.SendShowMetaReq(TSDB_MGMT_TABLE_DB, "");
-  CHECK_META("show databases", 18);
+  CHECK_META("show databases", 17);
   CHECK_SCHEMA(0, TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN - 1 + VARSTR_HEADER_SIZE, "name");
   CHECK_SCHEMA(1, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
   CHECK_SCHEMA(2, TSDB_DATA_TYPE_SMALLINT, 2, "vgroups");
@@ -45,7 +45,7 @@ TEST_F(MndTestDb, 01_ShowDb) {
   CHECK_SCHEMA(14, TSDB_DATA_TYPE_TINYINT, 1, "comp");
   CHECK_SCHEMA(15, TSDB_DATA_TYPE_TINYINT, 1, "cachelast");
   CHECK_SCHEMA(16, TSDB_DATA_TYPE_BINARY, 3 + VARSTR_HEADER_SIZE, "precision");
-  CHECK_SCHEMA(17, TSDB_DATA_TYPE_TINYINT, 1, "update");
+//  CHECK_SCHEMA(17, TSDB_DATA_TYPE_TINYINT, 1, "update");
 
   test.SendShowRetrieveReq();
   EXPECT_EQ(test.GetShowRows(), 0);
@@ -85,7 +85,7 @@ TEST_F(MndTestDb, 02_Create_Alter_Drop_Db) {
   }
 
   test.SendShowMetaReq(TSDB_MGMT_TABLE_DB, "");
-  CHECK_META("show databases", 18);
+  CHECK_META("show databases", 17);
 
   test.SendShowRetrieveReq();
   EXPECT_EQ(test.GetShowRows(), 1);
@@ -173,7 +173,7 @@ TEST_F(MndTestDb, 02_Create_Alter_Drop_Db) {
   test.Restart();
 
   test.SendShowMetaReq(TSDB_MGMT_TABLE_DB, "");
-  CHECK_META("show databases", 18);
+  CHECK_META("show databases", 17);
 
   test.SendShowRetrieveReq();
   EXPECT_EQ(test.GetShowRows(), 1);
@@ -215,7 +215,7 @@ TEST_F(MndTestDb, 02_Create_Alter_Drop_Db) {
   }
 
   test.SendShowMetaReq(TSDB_MGMT_TABLE_DB, "");
-  CHECK_META("show databases", 18);
+  CHECK_META("show databases", 17);
 
   test.SendShowRetrieveReq();
   EXPECT_EQ(test.GetShowRows(), 0);
@@ -255,7 +255,7 @@ TEST_F(MndTestDb, 03_Create_Use_Restart_Use_Db) {
   }
 
   test.SendShowMetaReq(TSDB_MGMT_TABLE_DB, "");
-  CHECK_META("show databases", 18);
+  CHECK_META("show databases", 17);
 
   test.SendShowRetrieveReq();
   EXPECT_EQ(test.GetShowRows(), 1);

source/dnode/mnode/impl/test/topic/topic.cpp

@@ -65,8 +65,7 @@ void* MndTestTopic::BuildCreateTopicReq(const char* topicName, const char* sql,
   strcpy(createReq.name, topicName);
   createReq.igExists = 0;
   createReq.sql = (char*)sql;
-  createReq.physicalPlan = (char*)"physicalPlan";
-  createReq.logicalPlan = (char*)"logicalPlan";
+  createReq.ast = NULL;
 
   int32_t contLen = tSerializeSCMCreateTopicReq(NULL, 0, &createReq);
   void*   pReq = rpcMallocCont(contLen);

source/dnode/mnode/sdb/src/sdbFile.c

@@ -196,7 +196,7 @@ int32_t sdbReadFile(SSdb *pSdb) {
     }
 
     int32_t totalLen = sizeof(SSdbRaw) + pRaw->dataLen + sizeof(int32_t);
-    if (!taosCheckChecksumWhole((const uint8_t *)pRaw, totalLen) != 0) {
+    if ((!taosCheckChecksumWhole((const uint8_t *)pRaw, totalLen)) != 0) {
       code = TSDB_CODE_CHECKSUM_ERROR;
       mError("failed to read file:%s since %s", file, tstrerror(code));
       break;

source/libs/catalog/src/catalog.c

@@ -1062,7 +1062,7 @@ int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
 
 
 int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size) {
-  int16_t widx = abs(id % mgmt->slotNum);
+  int16_t widx = abs((int)(id % mgmt->slotNum));
 
   SCtgRentSlot *slot = &mgmt->slots[widx];
   int32_t code = 0;
@@ -1092,11 +1092,11 @@ _return:
 }
 
 int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
-  int16_t widx = abs(id % mgmt->slotNum);
+  int16_t widx = abs((int)(id % mgmt->slotNum));
 
   SCtgRentSlot *slot = &mgmt->slots[widx];
   int32_t code = 0;
-  
+
   CTG_LOCK(CTG_WRITE, &slot->lock);
   if (NULL == slot->meta) {
     qError("empty meta slot, id:%"PRIx64", slot idx:%d, type:%d", id, widx, mgmt->type);
@@ -1133,7 +1133,7 @@ _return:
 }
 
 int32_t ctgMetaRentRemove(SCtgRentMgmt *mgmt, int64_t id, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
-  int16_t widx = abs(id % mgmt->slotNum);
+  int16_t widx = abs((int)(id % mgmt->slotNum));
 
   SCtgRentSlot *slot = &mgmt->slots[widx];
   int32_t code = 0;

source/libs/executor/inc/executil.h

@@ -50,7 +50,7 @@ typedef struct SGroupResInfo {
   int32_t totalGroup;
   int32_t currentGroup;
   int32_t index;
-  SArray* pRows;      // SArray<SResultRow*>
+  SArray* pRows;      // SArray<SResultRowPosition*>
   bool    ordered;
   int32_t position;
 } SGroupResInfo;
@@ -67,10 +67,15 @@ typedef struct SResultRow {
   char         *key;               // start key of current result row
 } SResultRow;
 
+typedef struct SResultRowPosition {
+  int32_t pageId;
+  int32_t offset;
+} SResultRowPosition;
+
 typedef struct SResultRowInfo {
-  SList* pRows;
-  SResultRow** pResult;    // result list
-//  int16_t      type:8;     // data type for hash key
+  SList       *pRows;
+  SResultRowPosition *pPosition;
+  SResultRow **pResult;    // result list
   int32_t      size;       // number of result set
   int32_t      capacity;   // max capacity
   int32_t      curPos;     // current active result row index of pResult list
@@ -131,7 +136,7 @@ static FORCE_INLINE char* getPosInResultPage_rv(SFilePage* page, int32_t rowOffs
   assert(rowOffset >= 0);
 
   int32_t numOfRows = 1;//(int32_t)getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery);
-  return ((char *)page->data) + rowOffset + offset * numOfRows;
+  return (char*) page + rowOffset + offset * numOfRows;
 }
 
 //bool isNullOperator(SColumnFilterElem *pFilter, const char* minval, const char* maxval, int16_t type);
@@ -139,12 +144,7 @@ static FORCE_INLINE char* getPosInResultPage_rv(SFilePage* page, int32_t rowOffs
 
 __filter_func_t getFilterOperator(int32_t lowerOptr, int32_t upperOptr);
 
-SResultRowPool* initResultRowPool(size_t size);
 SResultRow* getNewResultRow(SResultRowPool* p);
-int64_t getResultRowPoolMemSize(SResultRowPool* p);
-void* destroyResultRowPool(SResultRowPool* p);
-int32_t getNumOfAllocatedResultRows(SResultRowPool* p);
-int32_t getNumOfUsedResultRows(SResultRowPool* p);
 
 typedef struct {
   SArray* pResult;     // SArray<SResPair>

source/libs/executor/inc/executorimpl.h

@@ -240,12 +240,12 @@ typedef struct STaskAttr {
   SArray*         pUdfInfo;  // no need to free
 } STaskAttr;
 
-typedef int32_t (*__optr_open_fn_t)(void* param);
-typedef SSDataBlock* (*__optr_fn_t)(void* param, bool* newgroup);
-typedef void (*__optr_close_fn_t)(void* param, int32_t num);
-
 struct SOperatorInfo;
 
+typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* param);
+typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* param, bool* newgroup);
+typedef void (*__optr_close_fn_t)(void* param, int32_t num);
+
 typedef struct STaskIdInfo {
   uint64_t queryId;  // this is also a request id
   uint64_t subplanId;
@@ -275,36 +275,36 @@ typedef struct SExecTaskInfo {
 } SExecTaskInfo;
 
 typedef struct STaskRuntimeEnv {
-  jmp_buf    env;
-  STaskAttr* pQueryAttr;
-  uint32_t   status;  // query status
-  void*      qinfo;
-  uint8_t    scanFlag;  // denotes reversed scan of data or not
-  void*      pTsdbReadHandle;
-
-  int32_t              prevGroupId;  // previous executed group id
-  bool                 enableGroupData;
-  SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
-  SHashObj*            pResultRowHashTable;  // quick locate the window object for each result
-  SHashObj*            pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
-  SArray*              pResultRowArrayList;  // The array list that contains the Result rows
-  char*                keyBuf;               // window key buffer
+  jmp_buf         env;
+  STaskAttr*      pQueryAttr;
+  uint32_t        status;  // query status
+  void*           qinfo;
+  uint8_t         scanFlag;  // denotes reversed scan of data or not
+  void*           pTsdbReadHandle;
+
+  int32_t         prevGroupId;  // previous executed group id
+  bool            enableGroupData;
+  SDiskbasedBuf*  pResultBuf;           // query result buffer based on blocked-wised disk file
+  SHashObj*       pResultRowHashTable;  // quick locate the window object for each result
+  SHashObj*       pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
+  SArray*         pResultRowArrayList;  // The array list that contains the Result rows
+  char*           keyBuf;               // window key buffer
   // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
-  char** prevRow;
+  char**          prevRow;
   SResultRowPool* pool;
 
-  SArray*   prevResult;  // intermediate result, SArray<SInterResult>
-  STSBuf*   pTsBuf;      // timestamp filter list
-  STSCursor cur;
+  SArray*         prevResult;  // intermediate result, SArray<SInterResult>
+  STSBuf*         pTsBuf;      // timestamp filter list
+  STSCursor       cur;
 
-  char*                          tagVal;  // tag value of current data block
+  char*           tagVal;  // tag value of current data block
   struct SScalarFunctionSupport* scalarSup;
 
   SSDataBlock*    outputBuf;
   STableGroupInfo tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
   struct SOperatorInfo* proot;
-  SGroupResInfo         groupResInfo;
-  int64_t               currentOffset;  // dynamic offset value
+  SGroupResInfo   groupResInfo;
+  int64_t         currentOffset;  // dynamic offset value
 
   STableQueryInfo* current;
   SRspResultInfo   resultInfo;
@@ -328,7 +328,7 @@ typedef struct SOperatorInfo {
   char*                  name;          // name, used to show the query execution plan
   void*                  info;          // extension attribution
   SExprInfo*             pExpr;
-  STaskRuntimeEnv*       pRuntimeEnv;  // todo remove it
+  STaskRuntimeEnv*       pRuntimeEnv;   // todo remove it
   SExecTaskInfo*         pTaskInfo;
   SOperatorCostInfo      cost;
 
@@ -365,28 +365,6 @@ typedef struct SQInfo {
   STaskCostInfo   summary;
 } SQInfo;
 
-typedef struct STaskParam {
-  char*      sql;
-  char*      tagCond;
-  char*      colCond;
-  char*      tbnameCond;
-  char*      prevResult;
-  SArray*    pTableIdList;
-  SExprBasicInfo** pExpr;
-  SExprBasicInfo** pSecExpr;
-  SExprInfo* pExprs;
-  SExprInfo* pSecExprs;
-
-  SFilterInfo* pFilters;
-
-  SColIndex*       pGroupColIndex;
-  SColumnInfo*     pTagColumnInfo;
-  SGroupbyExpr*    pGroupbyExpr;
-  int32_t          tableScanOperator;
-  SArray*          pOperator;
-  struct SUdfInfo* pUdfInfo;
-} STaskParam;
-
 enum {
   EX_SOURCE_DATA_NOT_READY = 0x1,
   EX_SOURCE_DATA_READY     = 0x2,
@@ -472,75 +450,75 @@ typedef struct SSysTableScanInfo {
 } SSysTableScanInfo;
 
 typedef struct SOptrBasicInfo {
-  SResultRowInfo  resultRowInfo;
-  int32_t*        rowCellInfoOffset;  // offset value for each row result cell info
-  SqlFunctionCtx* pCtx;
-  SSDataBlock*    pRes;
-  int32_t         capacity;
+  SResultRowInfo     resultRowInfo;
+  int32_t*           rowCellInfoOffset;  // offset value for each row result cell info
+  SqlFunctionCtx*    pCtx;
+  SSDataBlock*       pRes;
+  int32_t            capacity;
 } SOptrBasicInfo;
 
 //TODO move the resultrowsiz together with SOptrBasicInfo:rowCellInfoOffset
 typedef struct SAggSupporter {
-  SHashObj*            pResultRowHashTable;  // quick locate the window object for each result
-  SHashObj*            pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
-  SArray*              pResultRowArrayList;  // The array list that contains the Result rows
-  char*                keyBuf;               // window key buffer
-  SResultRowPool      *pool;  // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
-  int32_t              resultRowSize;        // the result buffer size for each result row, with the meta data size for each row
+  SHashObj*          pResultRowHashTable;  // quick locate the window object for each result
+  SHashObj*          pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
+  SArray*            pResultRowArrayList;  // The array list that contains the Result rows
+  char*              keyBuf;               // window key buffer
+  SDiskbasedBuf     *pResultBuf;           // query result buffer based on blocked-wised disk file
+  int32_t            resultRowSize;        // the result buffer size for each result row, with the meta data size for each row
 } SAggSupporter;
 
 typedef struct STableIntervalOperatorInfo {
-  SOptrBasicInfo       binfo;
-  SDiskbasedBuf       *pResultBuf;           // query result buffer based on blocked-wised disk file
-  SGroupResInfo        groupResInfo;
-  SInterval            interval;
-  STimeWindow          win;
-  int32_t              precision;
-  bool                 timeWindowInterpo;
-  char               **pRow;
-  SAggSupporter        aggSup;
-  STableQueryInfo     *pCurrent;
-  int32_t              order;
+  SOptrBasicInfo     binfo;
+  SGroupResInfo      groupResInfo;
+  SInterval          interval;
+  STimeWindow        win;
+  int32_t            precision;
+  bool               timeWindowInterpo;
+  char             **pRow;
+  SAggSupporter      aggSup;
+  STableQueryInfo   *pCurrent;
+  int32_t            order;
 } STableIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
-  SOptrBasicInfo       binfo;
-  SDiskbasedBuf       *pResultBuf;           // query result buffer based on blocked-wised disk file
-  SAggSupporter        aggSup;
-  STableQueryInfo     *current;
-  uint32_t             groupId;
-  SGroupResInfo        groupResInfo;
-  STableQueryInfo     *pTableQueryInfo;
+  SOptrBasicInfo     binfo;
+  SDiskbasedBuf     *pResultBuf;           // query result buffer based on blocked-wised disk file
+  SAggSupporter      aggSup;
+  STableQueryInfo   *current;
+  uint32_t           groupId;
+  SGroupResInfo      groupResInfo;
+  STableQueryInfo   *pTableQueryInfo;
 } SAggOperatorInfo;
 
 typedef struct SProjectOperatorInfo {
-  SOptrBasicInfo binfo;
-  SSDataBlock* existDataBlock;
+  SOptrBasicInfo     binfo;
+  SSDataBlock       *existDataBlock;
+  int32_t            threshold;
+  bool               hasVarCol;
 } SProjectOperatorInfo;
 
 typedef struct SLimitOperatorInfo {
-  int64_t limit;
-  int64_t total;
+  SLimit             limit;
+  int64_t            currentOffset;
+  int64_t            currentRows;
 } SLimitOperatorInfo;
 
 typedef struct SSLimitOperatorInfo {
-  int64_t groupTotal;
-  int64_t currentGroupOffset;
-
-  int64_t rowsTotal;
-  int64_t currentOffset;
-  SLimit  limit;
-  SLimit  slimit;
-
-  char**       prevRow;
-  SArray*      orderColumnList;
-  bool         hasPrev;
-  bool         ignoreCurrentGroup;
-  bool         multigroupResult;
-  SSDataBlock* pRes;  // result buffer
-  SSDataBlock* pPrevBlock;
-  int64_t      capacity;
-  int64_t      threshold;
+  int64_t            groupTotal;
+  int64_t            currentGroupOffset;
+  int64_t            rowsTotal;
+  int64_t            currentOffset;
+  SLimit             limit;
+  SLimit             slimit;
+  char**             prevRow;
+  SArray*            orderColumnList;
+  bool               hasPrev;
+  bool               ignoreCurrentGroup;
+  bool               multigroupResult;
+  SSDataBlock*       pRes;  // result buffer
+  SSDataBlock*       pPrevBlock;
+  int64_t            capacity;
+  int64_t            threshold;
 } SSLimitOperatorInfo;
 
 typedef struct SFilterOperatorInfo {
@@ -563,14 +541,15 @@ typedef struct SGroupbyOperatorInfo {
   char*          prevData;  // previous group by value
 } SGroupbyOperatorInfo;
 
-typedef struct SSWindowOperatorInfo {
+typedef struct SSessionAggOperatorInfo {
   SOptrBasicInfo binfo;
+  SAggSupporter  aggSup;
   STimeWindow    curWindow;  // current time window
   TSKEY          prevTs;     // previous timestamp
   int32_t        numOfRows;  // number of rows
   int32_t        start;      // start row index
   bool           reptScan;   // next round scan
-} SSWindowOperatorInfo;
+} SSessionAggOperatorInfo;
 
 typedef struct SStateWindowOperatorInfo {
   SOptrBasicInfo binfo;
@@ -582,23 +561,6 @@ typedef struct SStateWindowOperatorInfo {
   bool           reptScan;
 } SStateWindowOperatorInfo;
 
-typedef struct SDistinctDataInfo {
-  int32_t index;
-  int32_t type;
-  int32_t bytes;
-} SDistinctDataInfo;
-
-typedef struct SDistinctOperatorInfo {
-  SHashObj*    pSet;
-  SSDataBlock* pRes;
-  bool         recordNullVal;  // has already record the null value, no need to try again
-  int64_t      threshold;
-  int64_t      outputCapacity;
-  int32_t      totalBytes;
-  char*        buf;
-  SArray*      pDistinctDataInfo;
-} SDistinctOperatorInfo;
-
 typedef struct SSortedMergeOperatorInfo {
   SOptrBasicInfo     binfo;
   bool               hasVarCol;
@@ -624,45 +586,61 @@ typedef struct SSortedMergeOperatorInfo {
 } SSortedMergeOperatorInfo;
 
 typedef struct SOrderOperatorInfo {
-  uint32_t                sortBufSize;  // max buffer size for in-memory sort
-  SSDataBlock            *pDataBlock;
-  bool                    hasVarCol;    // has variable length column, such as binary/varchar/nchar
-  SArray                 *orderInfo;
-  bool                    nullFirst;
-  SSortHandle            *pSortHandle;
-
-  int32_t                 bufPageSize;
-  int32_t                 numOfRowsInRes;
+  uint32_t           sortBufSize;  // max buffer size for in-memory sort
+  SSDataBlock       *pDataBlock;
+  bool               hasVarCol;    // has variable length column, such as binary/varchar/nchar
+  SArray            *orderInfo;
+  bool               nullFirst;
+  SSortHandle       *pSortHandle;
+  int32_t            bufPageSize;
+  int32_t            numOfRowsInRes;
 
   // TODO extact struct
-  int64_t                 startTs;       // sort start time
-  uint64_t                sortElapsed;   // sort elapsed time, time to flush to disk not included.
-  uint64_t                totalSize;     // total load bytes from remote
-  uint64_t                totalRows;     // total number of rows
-  uint64_t                totalElapsed;  // total elapsed time
+  int64_t            startTs;       // sort start time
+  uint64_t           sortElapsed;   // sort elapsed time, time to flush to disk not included.
+  uint64_t           totalSize;     // total load bytes from remote
+  uint64_t           totalRows;     // total number of rows
+  uint64_t           totalElapsed;  // total elapsed time
 } SOrderOperatorInfo;
 
+typedef struct SDistinctDataInfo {
+  int32_t index;
+  int32_t type;
+  int32_t bytes;
+} SDistinctDataInfo;
+
+typedef struct SDistinctOperatorInfo {
+  SHashObj*    pSet;
+  SSDataBlock* pRes;
+  bool         recordNullVal;  // has already record the null value, no need to try again
+  int64_t      threshold;
+  int64_t      outputCapacity;
+  int32_t      totalBytes;
+  char*        buf;
+  SArray*      pDistinctDataInfo;
+} SDistinctOperatorInfo;
+
 SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfOutput,
                                            int32_t repeatTime, int32_t reverseTime, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
-SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SSDataBlock* pResultBlock, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
-SOperatorInfo* createMultiTableAggOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SSDataBlock* pResultBlock, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
-SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createOrderOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SArray* pOrderVal, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SArray* pExprInfo, SArray* pOrderVal, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
+                                           SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
+SOperatorInfo* createMultiTableAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
+SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createOrderOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SArray* pOrderVal, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pOrderVal, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, const SArray* pExprInfo, const SSchema* pSchema,
                                               int32_t tableType, SEpSet epset, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createLimitOperatorInfo(SOperatorInfo* downstream, int32_t numOfDownstream, SLimit* pLimit, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream);
-SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SInterval* pInterval, SExecTaskInfo* pTaskInfo);
-
-SOperatorInfo* createLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream);
+SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval,
+                                          const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
 
+SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
 SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
                                                  SExprInfo* pExpr, int32_t numOfOutput);
-SOperatorInfo* createSWindowOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
-                                         int32_t numOfOutput);
+
 SOperatorInfo* createFillOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
                                       int32_t numOfOutput, bool multigroupResult);
 SOperatorInfo* createGroupbyOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
@@ -699,16 +677,13 @@ SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numO
 void* doDestroyFilterInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols);
 
 void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
-void finalizeQueryResult(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SResultRowInfo* pResultRowInfo,
-                         int32_t* rowCellInfoOffset);
+void finalizeQueryResult(SqlFunctionCtx* pCtx, int32_t numOfOutput);
+
 void clearOutputBuf(SOptrBasicInfo* pBInfo, int32_t* bufCapacity);
 void copyTsColoum(SSDataBlock* pRes, SqlFunctionCtx* pCtx, int32_t numOfOutput);
 
 int32_t createQueryFilter(char* data, uint16_t len, SFilterInfo** pFilters);
 
-int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, void* sourceOptr, SQInfo* pQInfo, STaskParam* param, char* start,
-                  int32_t prevResultLen, void* merger);
-
 int32_t createFilterInfo(STaskAttr* pQueryAttr, uint64_t qId);
 void    freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters);
 

source/libs/executor/src/dataDispatcher.c

@@ -115,7 +115,8 @@ static bool allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput,
     return false;
   }
 
-  pBuf->allocSize = sizeof(SRetrieveTableRsp) + pDispatcher->pSchema->resultRowSize * pInput->pData->info.rows;
+  // struct size + data payload + length for each column
+  pBuf->allocSize = sizeof(SRetrieveTableRsp) + pDispatcher->pSchema->resultRowSize * pInput->pData->info.rows + pInput->pData->info.numOfCols * sizeof(int32_t);
   pBuf->pData = malloc(pBuf->allocSize);
   if (pBuf->pData == NULL) {
     qError("SinkNode failed to malloc memory, size:%d, code:%d", pBuf->allocSize, TAOS_SYSTEM_ERROR(errno));

source/libs/executor/src/dataSinkMgt.c

@@ -23,6 +23,7 @@ static SDataSinkManager gDataSinkManager = {0};
 int32_t dsDataSinkMgtInit(SDataSinkMgtCfg *cfg) {
   gDataSinkManager.cfg = *cfg;
   pthread_mutex_init(&gDataSinkManager.mutex, NULL);
+  return 0; // to avoid compiler eror
 }
 
 int32_t dsCreateDataSinker(const SDataSinkNode *pDataSink, DataSinkHandle* pHandle) {

source/libs/executor/src/executil.c

@@ -59,7 +59,8 @@ int32_t initResultRowInfo(SResultRowInfo *pResultRowInfo, int32_t size) {
   pResultRowInfo->capacity = size;
 
   pResultRowInfo->pResult = calloc(pResultRowInfo->capacity, POINTER_BYTES);
-  if (pResultRowInfo->pResult == NULL) {
+  pResultRowInfo->pPosition = calloc(pResultRowInfo->capacity, sizeof(SResultRowPosition));
+  if (pResultRowInfo->pResult == NULL || pResultRowInfo->pPosition == NULL) {
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
 
@@ -182,22 +183,6 @@ size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
   return rowSize;
 }
 
-SResultRowPool* initResultRowPool(size_t size) {
-  SResultRowPool* p = calloc(1, sizeof(SResultRowPool));
-  if (p == NULL) {
-    return NULL;
-  }
-
-  p->numOfElemPerBlock = 128;
-
-  p->elemSize = (int32_t) size;
-  p->blockSize = p->numOfElemPerBlock * p->elemSize;
-  p->position.pos = 0;
-
-  p->pData = taosArrayInit(8, POINTER_BYTES);
-  return p;
-}
-
 SResultRow* getNewResultRow(SResultRowPool* p) {
   if (p == NULL) {
     return NULL;
@@ -221,132 +206,6 @@ SResultRow* getNewResultRow(SResultRowPool* p) {
   return ptr;
 }
 
-int64_t getResultRowPoolMemSize(SResultRowPool* p) {
-  if (p == NULL) {
-    return 0;
-  }
-
-  return taosArrayGetSize(p->pData) * p->blockSize;
-}
-
-int32_t getNumOfAllocatedResultRows(SResultRowPool* p) {
-  return (int32_t) taosArrayGetSize(p->pData) * p->numOfElemPerBlock;
-}
-
-int32_t getNumOfUsedResultRows(SResultRowPool* p) {
-  return getNumOfAllocatedResultRows(p) - p->numOfElemPerBlock + p->position.pos;
-}
-
-void* destroyResultRowPool(SResultRowPool* p) {
-  if (p == NULL) {
-    return NULL;
-  }
-
-  size_t size = taosArrayGetSize(p->pData);
-  for(int32_t i = 0; i < size; ++i) {
-    void** ptr = taosArrayGet(p->pData, i);
-    tfree(*ptr);
-  }
-
-  taosArrayDestroy(p->pData);
-
-  tfree(p);
-  return NULL;
-}
-
-void interResToBinary(SBufferWriter* bw, SArray* pRes, int32_t tagLen) {
-  uint32_t numOfGroup = (uint32_t) taosArrayGetSize(pRes);
-  tbufWriteUint32(bw, numOfGroup);
-  tbufWriteUint16(bw, tagLen);
-
-  for(int32_t i = 0; i < numOfGroup; ++i) {
-    SInterResult* pOne = taosArrayGet(pRes, i);
-    if (tagLen > 0) {
-      tbufWriteBinary(bw, pOne->tags, tagLen);
-    }
-
-    uint32_t numOfCols = (uint32_t) taosArrayGetSize(pOne->pResult);
-    tbufWriteUint32(bw, numOfCols);
-    for(int32_t j = 0; j < numOfCols; ++j) {
-      SStddevInterResult* p = taosArrayGet(pOne->pResult, j);
-      uint32_t numOfRows = (uint32_t) taosArrayGetSize(p->pResult);
-
-      tbufWriteUint16(bw, p->colId);
-      tbufWriteUint32(bw, numOfRows);
-
-      for(int32_t k = 0; k < numOfRows; ++k) {
-//        SResPair v = *(SResPair*) taosArrayGet(p->pResult, k);
-//        tbufWriteDouble(bw, v.avg);
-//        tbufWriteInt64(bw, v.key);
-      }
-    }
-  }
-}
-
-SArray* interResFromBinary(const char* data, int32_t len) {
-  SBufferReader br = tbufInitReader(data, len, false);
-  uint32_t numOfGroup = tbufReadUint32(&br);
-  uint16_t tagLen = tbufReadUint16(&br);
-
-  char* tag = NULL;
-  if (tagLen > 0) {
-    tag = calloc(1, tagLen);
-  }
-
-  SArray* pResult = taosArrayInit(4, sizeof(SInterResult));
-
-  for(int32_t i = 0; i < numOfGroup; ++i) {
-    if (tagLen > 0) {
-      memset(tag, 0, tagLen);
-      tbufReadToBinary(&br, tag, tagLen);
-    }
-
-    uint32_t numOfCols = tbufReadUint32(&br);
-
-    SArray* p = taosArrayInit(numOfCols, sizeof(SStddevInterResult));
-    for(int32_t j = 0; j < numOfCols; ++j) {
-//      int16_t colId = tbufReadUint16(&br);
-      int32_t numOfRows = tbufReadUint32(&br);
-
-//      SStddevInterResult interRes = {.colId = colId, .pResult = taosArrayInit(4, sizeof(struct SResPair)),};
-      for(int32_t k = 0; k < numOfRows; ++k) {
-//        SResPair px = {0};
-//        px.avg = tbufReadDouble(&br);
-//        px.key = tbufReadInt64(&br);
-//
-//        taosArrayPush(interRes.pResult, &px);
-      }
-
-//      taosArrayPush(p, &interRes);
-    }
-
-    char* p1 = NULL;
-    if (tagLen > 0) {
-      p1 = malloc(tagLen);
-      memcpy(p1, tag, tagLen);
-    }
-
-    SInterResult d = {.pResult = p, .tags = p1,};
-    taosArrayPush(pResult, &d);
-  }
-
-  tfree(tag);
-  return pResult;
-}
-
-void freeInterResult(void* param) {
-  SInterResult* pResult = (SInterResult*) param;
-  tfree(pResult->tags);
-
-  int32_t numOfCols = (int32_t) taosArrayGetSize(pResult->pResult);
-  for(int32_t i = 0; i < numOfCols; ++i) {
-    SStddevInterResult *p = taosArrayGet(pResult->pResult, i);
-    taosArrayDestroy(p->pResult);
-  }
-
-  taosArrayDestroy(pResult->pResult);
-}
-
 void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
   assert(pGroupResInfo != NULL);
 
@@ -360,7 +219,7 @@ void initGroupResInfo(SGroupResInfo* pGroupResInfo, SResultRowInfo* pResultInfo)
     taosArrayDestroy(pGroupResInfo->pRows);
   }
 
-  pGroupResInfo->pRows = taosArrayFromList(pResultInfo->pResult, pResultInfo->size, POINTER_BYTES);
+  pGroupResInfo->pRows = taosArrayFromList(pResultInfo->pPosition, pResultInfo->size, sizeof(SResultRowPosition));
   pGroupResInfo->index = 0;
   assert(pGroupResInfo->index <= getNumOfTotalRes(pGroupResInfo));
 }

source/libs/executor/src/tlinearhash.c

@@ -413,7 +413,7 @@ int32_t tHashRemove(SLHashObj* pHashObj, const void *key, size_t keyLen) {
 
 void tHashPrint(const SLHashObj* pHashObj, int32_t type) {
   printf("==================== linear hash ====================\n");
-  printf("total bucket:%d, size:%ld, ratio:%.2f\n", pHashObj->numOfBuckets, pHashObj->size, LHASH_CAP_RATIO);
+  printf("total bucket:%d, size:%" PRId64 ", ratio:%.2f\n", pHashObj->numOfBuckets, pHashObj->size, LHASH_CAP_RATIO);
 
   dBufSetPrintInfo(pHashObj->pBuf);
 
@@ -425,4 +425,4 @@ void tHashPrint(const SLHashObj* pHashObj, int32_t type) {
   } else {
     dBufPrintStatis(pHashObj->pBuf);
   }
-}
+}
\ No newline at end of file

source/libs/executor/src/tsort.c

@@ -37,7 +37,6 @@ typedef struct SSortHandle {
 
   SArray           *pOrderInfo;
   bool              nullFirst;
-  bool              hasVarCol;
   SArray           *pOrderedSource;
 
   _sort_fetch_block_fn_t  fetchfp;
@@ -77,6 +76,10 @@ static SSDataBlock* createDataBlock_rv(SSchema* pSchema, int32_t numOfCols) {
     colInfo.info.bytes = pSchema[i].bytes;
     colInfo.info.colId = pSchema[i].colId;
     taosArrayPush(pBlock->pDataBlock, &colInfo);
+
+    if (IS_VAR_DATA_TYPE(colInfo.info.type)) {
+      pBlock->info.hasVarCol = true;
+    }
   }
 
   return pBlock;
@@ -155,7 +158,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
 
   while(start < pDataBlock->info.rows) {
     int32_t stop = 0;
-    blockDataSplitRows(pDataBlock, pHandle->hasVarCol, start, &stop, pHandle->pageSize);
+    blockDataSplitRows(pDataBlock, pDataBlock->info.hasVarCol, start, &stop, pHandle->pageSize);
     SSDataBlock* p = blockDataExtractBlock(pDataBlock, start, stop - start + 1);
     if (p == NULL) {
       return terrno;
@@ -179,7 +182,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
     start = stop + 1;
   }
 
-  blockDataClearup(pDataBlock, pHandle->hasVarCol);
+  blockDataClearup(pDataBlock);
 
   SSDataBlock* pBlock = createOneDataBlock(pDataBlock);
   int32_t code = doAddNewExternalMemSource(pHandle->pBuf, pHandle->pOrderedSource, pBlock, &pHandle->sourceId);
@@ -309,7 +312,7 @@ static int32_t adjustMergeTreeForNextTuple(SExternalMemSource *pSource, SMultiwa
 }
 
 static SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SMsortComparParam* cmpParam, int32_t capacity) {
-  blockDataClearup(pHandle->pDataBlock, pHandle->hasVarCol);
+  blockDataClearup(pHandle->pDataBlock);
 
   while(1) {
     if (cmpParam->numOfSources == pHandle->numOfCompletedSources) {
@@ -475,7 +478,7 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
         setBufPageDirty(pPage, true);
         releaseBufPage(pHandle->pBuf, pPage);
 
-        blockDataClearup(pDataBlock, pHandle->hasVarCol);
+        blockDataClearup(pDataBlock);
       }
 
       tMergeTreeDestroy(pHandle->pMergeTree);

source/libs/executor/test/executorTests.cpp

@@ -55,8 +55,7 @@ typedef struct SDummyInputInfo {
   SSDataBlock* pBlock;
 } SDummyInputInfo;
 
-SSDataBlock* getDummyBlock(void* param, bool* newgroup) {
-  SOperatorInfo* pOperator = static_cast<SOperatorInfo*>(param);
+SSDataBlock* getDummyBlock(SOperatorInfo* pOperator, bool* newgroup) {
   SDummyInputInfo* pInfo = static_cast<SDummyInputInfo*>(pOperator->info);
   if (pInfo->current >= pInfo->totalPages) {
     return NULL;
@@ -87,7 +86,7 @@ SSDataBlock* getDummyBlock(void* param, bool* newgroup) {
 //
 //    taosArrayPush(pInfo->pBlock->pDataBlock, &colInfo1);
   } else {
-    blockDataClearup(pInfo->pBlock, true);
+    blockDataClearup(pInfo->pBlock);
   }
 
   SSDataBlock* pBlock = pInfo->pBlock;
@@ -122,8 +121,7 @@ SSDataBlock* getDummyBlock(void* param, bool* newgroup) {
   return pBlock;
 }
 
-SSDataBlock* get2ColsDummyBlock(void* param, bool* newgroup) {
-  SOperatorInfo* pOperator = static_cast<SOperatorInfo*>(param);
+SSDataBlock* get2ColsDummyBlock(SOperatorInfo* pOperator, bool* newgroup) {
   SDummyInputInfo* pInfo = static_cast<SDummyInputInfo*>(pOperator->info);
   if (pInfo->current >= pInfo->totalPages) {
     return NULL;
@@ -153,7 +151,7 @@ SSDataBlock* get2ColsDummyBlock(void* param, bool* newgroup) {
 
     taosArrayPush(pInfo->pBlock->pDataBlock, &colInfo1);
   } else {
-    blockDataClearup(pInfo->pBlock, false);
+    blockDataClearup(pInfo->pBlock);
   }
 
   SSDataBlock* pBlock = pInfo->pBlock;

source/libs/function/inc/taggfunction.h

@@ -46,13 +46,6 @@ extern SAggFunctionInfo aggFunc[35];
 #define DATA_SET_FLAG ','  // to denote the output area has data, not null value
 #define DATA_SET_FLAG_SIZE sizeof(DATA_SET_FLAG)
 
-#define TOP_BOTTOM_QUERY_LIMIT   100
-
-#define QUERY_IS_STABLE_QUERY(type)      (((type)&TSDB_QUERY_TYPE_STABLE_QUERY) != 0)
-#define QUERY_IS_JOIN_QUERY(type)        (TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_JOIN_QUERY))
-#define QUERY_IS_PROJECTION_QUERY(type)  (((type)&TSDB_QUERY_TYPE_PROJECTION_QUERY) != 0)
-#define QUERY_IS_FREE_RESOURCE(type)     (((type)&TSDB_QUERY_TYPE_FREE_RESOURCE) != 0)
-
 typedef struct SInterpInfoDetail {
   TSKEY  ts;  // interp specified timestamp
   int8_t type;
@@ -61,9 +54,6 @@ typedef struct SInterpInfoDetail {
 
 #define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowEntryInfo)))
 
-#define IS_STREAM_QUERY_VALID(x)  (((x)&TSDB_FUNCSTATE_STREAM) != 0)
-#define IS_MULTIOUTPUT(x)         (((x)&TSDB_FUNCSTATE_MO) != 0)
-
 typedef struct STwaInfo {
   int8_t      hasResult;  // flag to denote has value
   double      dOutput;
@@ -71,8 +61,6 @@ typedef struct STwaInfo {
   STimeWindow win;
 } STwaInfo;
 
-extern int32_t functionCompatList[]; // compatible check array list
-
 bool topbot_datablock_filter(SqlFunctionCtx *pCtx, const char *minval, const char *maxval);
 
 /**

source/libs/function/src/builtinsimpl.c

@@ -52,10 +52,6 @@ static void doFinalizer(SResultRowEntryInfo* pResInfo) { cleanupResultRowEntry(p
 
 void functionFinalizer(SqlFunctionCtx *pCtx) {
   SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
-  if (pResInfo->hasResult != DATA_SET_FLAG) {
-//    setNull(pCtx->pOutput, pCtx->resDataInfo.type, pCtx->resDataInfo.bytes);
-  }
-
   doFinalizer(pResInfo);
 }
 
@@ -398,7 +394,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx *pCtx, int32_t isMinFunc) {
       int32_t *pData = (int32_t*)pCol->pData;
       int32_t *val = (int32_t*) buf;
 
-      for (int32_t i = 0; i < pCtx->size; ++i) {
+      for (int32_t i = start; i < start + numOfRows; ++i) {
         if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
           continue;
         }

source/libs/function/src/tfunction.c

-#include "os.h"
-#include "tarray.h"
-#include "function.h"
-#include "thash.h"
-#include "taggfunction.h"
-
-static SHashObj* functionHashTable = NULL;
-static SHashObj* udfHashTable = NULL;
-
-static void doInitFunctionHashTable() {
-  int numOfEntries = tListLen(aggFunc);
-  functionHashTable = taosHashInit(numOfEntries, MurmurHash3_32, false, false);
-  for (int32_t i = 0; i < numOfEntries; i++) {
-    int32_t len = (uint32_t)strlen(aggFunc[i].name);
-
-    SAggFunctionInfo* ptr = &aggFunc[i];
-    taosHashPut(functionHashTable, aggFunc[i].name, len, (void*)&ptr, POINTER_BYTES);
-  }
-
-/*
-  numOfEntries = tListLen(scalarFunc);
-  for(int32_t i = 0; i < numOfEntries; ++i) {
-    int32_t len = (int32_t) strlen(scalarFunc[i].name);
-    SScalarFunctionInfo* ptr = &scalarFunc[i];
-    taosHashPut(functionHashTable, scalarFunc[i].name, len, (void*)&ptr, POINTER_BYTES);
-  }
-*/
-
-  udfHashTable = taosHashInit(numOfEntries, MurmurHash3_32, true, true);
-}
-
-static pthread_once_t functionHashTableInit = PTHREAD_ONCE_INIT;
-
-int32_t qIsBuiltinFunction(const char* name, int32_t len, bool* scalarFunction) {
-  pthread_once(&functionHashTableInit, doInitFunctionHashTable);
-
-  SAggFunctionInfo** pInfo = taosHashGet(functionHashTable, name, len);
-  if (pInfo != NULL) {
-    *scalarFunction = ((*pInfo)->type == FUNCTION_TYPE_SCALAR);
-    return (*pInfo)->functionId;
-  } else {
-    return -1;
-  }
-}
-
-bool qIsValidUdf(SArray* pUdfInfo, const char* name, int32_t len, int32_t* functionId) {
-  return true;
-}
-
-bool qIsAggregateFunction(const char* functionName) {
-  assert(functionName != NULL);
-  bool scalarfunc = false;
-  qIsBuiltinFunction(functionName, strlen(functionName), &scalarfunc);
-
-  return !scalarfunc;
-}
-
-bool qIsSelectivityFunction(const char* functionName) {
-  assert(functionName != NULL);
-  pthread_once(&functionHashTableInit, doInitFunctionHashTable);
-
-  size_t len = strlen(functionName);
-  SAggFunctionInfo** pInfo = taosHashGet(functionHashTable, functionName, len);
-  if (pInfo != NULL) {
-    return ((*pInfo)->status | FUNCSTATE_SELECTIVITY) != 0;
-  }
-
-  return false;
-}
-
-SAggFunctionInfo* qGetFunctionInfo(const char* name, int32_t len) {
-  pthread_once(&functionHashTableInit, doInitFunctionHashTable);
-
-  SAggFunctionInfo** pInfo = taosHashGet(functionHashTable, name, len);
-  if (pInfo != NULL) {
-    return (*pInfo);
-  } else {
-    return NULL;
-  }
-}
-
-void qAddUdfInfo(uint64_t id, SUdfInfo* pUdfInfo) {
-  int32_t len = (uint32_t)strlen(pUdfInfo->name);
-  taosHashPut(udfHashTable, pUdfInfo->name, len, (void*)&pUdfInfo, POINTER_BYTES);
-}
-
-void qRemoveUdfInfo(uint64_t id, SUdfInfo* pUdfInfo) {
-  int32_t len = (uint32_t)strlen(pUdfInfo->name);
-  taosHashRemove(udfHashTable, pUdfInfo->name, len);
-}
-
-bool isTagsQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    char* f = *(char**) taosArrayGet(pFunctionIdList, i);
-
-    // todo handle count(tbname) query
-    if (strcmp(f, "project") != 0 && strcmp(f, "count") != 0) {
-      return false;
-    }
-
-    // "select count(tbname)" query
-//    if (functId == FUNCTION_COUNT && pExpr->base.colpDesc->colId == TSDB_TBNAME_COLUMN_INDEX) {
-//      continue;
-//    }
-  }
-
-  return true;
-}
-
-//bool tscMultiRoundQuery(SArray* pFunctionIdList, int32_t index) {
-//  if (!UTIL_TABLE_IS_SUPER_TABLE(pQueryInfo->pTableMetaInfo[index])) {
-//    return false;
-//  }
-//
-//  size_t numOfExprs = (int32_t) getNumOfExprs(pQueryInfo);
-//  for(int32_t i = 0; i < numOfExprs; ++i) {
-//    SExprInfo* pExpr = getExprInfo(pQueryInfo, i);
-//    if (pExpr->base.functionId == FUNCTION_STDDEV_DST) {
-//      return true;
-//    }
-//  }
-//
-//  return false;
-//}
-
-bool isProjectionQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    char* f = *(char**) taosArrayGet(pFunctionIdList, i);
-    if (strcmp(f, "project") == 0) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool isDiffDerivativeQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-    if (f == FUNCTION_TS_DUMMY) {
-      continue;
-    }
-
-    if (f == FUNCTION_DIFF || f == FUNCTION_DERIVATIVE) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool isInterpQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-    if (f == FUNCTION_TAG || f == FUNCTION_TS) {
-      continue;
-    }
-
-    if (f != FUNCTION_INTERP) {
-      return false;
-    }
-  }
-
-  return true;
-}
-
-bool isArithmeticQueryOnAggResult(SArray* pFunctionIdList) {
-  if (isProjectionQuery(pFunctionIdList)) {
-    return false;
-  }
-
-  assert(0);
-
-//  size_t numOfOutput = getNumOfFields(pQueryInfo);
-//  for(int32_t i = 0; i < numOfOutput; ++i) {
-//    SExprInfo* pExprInfo = tscFieldInfoGetInternalField(&pQueryInfo->fieldsInfo, i)->pExpr;
-//    if (pExprInfo->pExpr != NULL) {
-//      return true;
-//    }
-//  }
-
-  return false;
-}
-
-bool isGroupbyColumn(SGroupbyExpr* pGroupby) {
-  return !pGroupby->groupbyTag;
-}
-
-bool isTopBotQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    char* f = *(char**) taosArrayGet(pFunctionIdList, i);
-    if (strcmp(f, "project") == 0) {
-      continue;
-    }
-
-    if (strcmp(f, "top") == 0 || strcmp(f, "bottom") == 0) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool isTsCompQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  if (num != 1) {
-    return false;
-  }
-
-  int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, 0);
-  return f == FUNCTION_TS_COMP;
-}
-
-bool isTWAQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-    if (f == FUNCTION_TWA) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool isIrateQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-    if (f == FUNCTION_IRATE) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool isStabledev(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-    if (f == FUNCTION_STDDEV_DST) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool needReverseScan(SArray* pFunctionIdList) {
-  assert(0);
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < num; ++i) {
-    int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-    if (f == FUNCTION_TS || f == FUNCTION_TS_DUMMY || f == FUNCTION_TAG) {
-      continue;
-    }
-
-//    if ((f == FUNCTION_FIRST || f == FUNCTION_FIRST_DST) && pQueryInfo->order.order == TSDB_ORDER_DESC) {
-//      return true;
-//    }
-
-    if (f == FUNCTION_LAST || f == FUNCTION_LAST_DST) {
-      // the scan order to acquire the last result of the specified column
-//      int32_t order = (int32_t)pExpr->base.param[0].i64;
-//      if (order != pQueryInfo->order.order) {
-//        return true;
-//      }
-    }
-  }
-
-  return false;
-}
-
-bool isAgg(SArray* pFunctionIdList) {
-  size_t size = taosArrayGetSize(pFunctionIdList);
-  for (int32_t i = 0; i < size; ++i) {
-    char* f = *(char**) taosArrayGet(pFunctionIdList, i);
-    if (strcmp(f, "project") == 0) {
-      return false;
-    }
-
-    if (qIsAggregateFunction(f)) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-bool isBlockDistQuery(SArray* pFunctionIdList) {
-  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-  char* f = *(char**) taosArrayGet(pFunctionIdList, 0);
-  return (num == 1 && strcmp(f, "block_dist") == 0);
-}
-
-bool isTwoStageSTableQuery(SArray* pFunctionIdList, int32_t tableIndex) {
-//  if (pQueryInfo == NULL) {
-//    return false;
-//  }
-//
-//  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, tableIndex);
-//  if (pTableMetaInfo == NULL) {
-//    return false;
-//  }
-//
-//  if ((pQueryInfo->type & TSDB_QUERY_TYPE_FREE_RESOURCE) == TSDB_QUERY_TYPE_FREE_RESOURCE) {
-//    return false;
-//  }
-//
-//  // for ordered projection query, iterate all qualified vnodes sequentially
-//  if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, tableIndex)) {
-//    return false;
-//  }
-//
-//  if (!TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_STABLE_SUBQUERY) && pQueryInfo->command == TSDB_SQL_SELECT) {
-//    return UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo);
-//  }
-
-  return false;
-}
-
-bool isProjectionQueryOnSTable(SArray* pFunctionIdList, int32_t tableIndex) {
-//  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, tableIndex);
-//
-//  /*
-//   * In following cases, return false for non ordered project query on super table
-//   * 1. failed to get tableMeta from server; 2. not a super table; 3. limitation is 0;
-//   * 4. show queries, instead of a select query
-//   */
-//  size_t numOfExprs = getNumOfExprs(pQueryInfo);
-//  if (pTableMetaInfo == NULL || !UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo) ||
-//      pQueryInfo->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT || numOfExprs == 0) {
-//    return false;
-//  }
-//
-//  for (int32_t i = 0; i < numOfExprs; ++i) {
-//    int32_t functionId = getExprInfo(pQueryInfo, i)->base.functionId;
-//
-//    if (functionId < 0) {
-//      SUdfInfo* pUdfInfo = taosArrayGet(pQueryInfo->pUdfInfo, -1 * functionId - 1);
-//      if (pUdfInfo->funcType == TSDB_FUNC_TYPE_AGGREGATE) {
-//        return false;
-//      }
-//
-//      continue;
-//    }
-//
-//    if (functionId != FUNCTION_PRJ &&
-//        functionId != FUNCTION_TAGPRJ &&
-//        functionId != FUNCTION_TAG &&
-//        functionId != FUNCTION_TS &&
-//        functionId != FUNCTION_ARITHM &&
-//        functionId != FUNCTION_TS_COMP &&
-//        functionId != FUNCTION_DIFF &&
-//        functionId != FUNCTION_DERIVATIVE &&
-//        functionId != FUNCTION_TS_DUMMY &&
-//        functionId != FUNCTION_TID_TAG) {
-//      return false;
-//    }
-//  }
-
-  return true;
-}
-
-bool hasTagValOutput(SArray* pFunctionIdList) {
-  size_t size = taosArrayGetSize(pFunctionIdList);
-
-  //  if (numOfExprs == 1 && pExpr1->base.functionId == FUNCTION_TS_COMP) {
-//    return true;
-//  }
-
-  for (int32_t i = 0; i < size; ++i) {
-    int32_t functionId = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-
-    // ts_comp column required the tag value for join filter
-    if (functionId == FUNCTION_TAG || functionId == FUNCTION_TAGPRJ) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
-//bool timeWindowInterpoRequired(SArray* pFunctionIdList) {
-//  int32_t num = (int32_t) taosArrayGetSize(pFunctionIdList);
-//  for (int32_t i = 0; i < num; ++i) {
-//    int32_t f = *(int16_t*) taosArrayGet(pFunctionIdList, i);
-//    if (f == FUNCTION_TWA || f == FUNCTION_INTERP) {
-//      return true;
-//    }
-//  }
-//
-//  return false;
-//}
-
-void extractFunctionDesc(SArray* pFunctionIdList, SMultiFunctionsDesc* pDesc) {
-  assert(pFunctionIdList != NULL);
-
-  pDesc->blockDistribution = isBlockDistQuery(pFunctionIdList);
-  if (pDesc->blockDistribution) {
-    return;
-  }
-
-//  pDesc->projectionQuery = isProjectionQuery(pFunctionIdList);
-//  pDesc->onlyTagQuery    = isTagsQuery(pFunctionIdList);
-  pDesc->interpQuery     = isInterpQuery(pFunctionIdList);
-  pDesc->topbotQuery     = isTopBotQuery(pFunctionIdList);
-  pDesc->agg             = isAgg(pFunctionIdList);
-}

source/libs/index/src/index.c

@@ -240,6 +240,7 @@ int indexRebuild(SIndex* index, SIndexOpts* opts){
 #ifdef USE_INVERTED_INDEX
 #endif
 
+    return 0;
 }
 
 SIndexOpts* indexOptsCreate() {

source/libs/index/src/index_cache.c

@@ -190,7 +190,10 @@ int indexCacheSchedToMerge(IndexCache* pCache) {
   schedMsg.msg = NULL;
 
   taosScheduleTask(indexQhandle, &schedMsg);
+
+  return 0;
 }
+
 static void indexCacheMakeRoomForWrite(IndexCache* cache) {
   while (true) {
     if (cache->occupiedMem * MEM_ESTIMATE_RADIO < MEM_THRESHOLD) {

source/libs/index/src/index_fst.c

@@ -571,6 +571,8 @@ uint64_t fstStateFindInput(FstState* s, FstNode* node, uint8_t b, bool* null) {
     }
     fstSliceDestroy(&t);
   }
+
+  return 0;
 }
 
 // fst node function
@@ -1027,6 +1029,8 @@ Fst* fstCreate(FstSlice* slice) {
 FST_CREAT_FAILED:
   free(fst->meta);
   free(fst);
+
+  return NULL;
 }
 void fstDestroy(Fst* fst) {
   if (fst) {
@@ -1286,6 +1290,8 @@ bool streamWithStateSeekMin(StreamWithState* sws, FstBoundWithData* min) {
     }
     return false;
   }
+
+  return false;
 }
 
 StreamWithStateResult* streamWithStateNextWith(StreamWithState* sws, StreamCallback callback) {

source/libs/nodes/src/nodesCodeFuncs.c

@@ -151,8 +151,7 @@ static int32_t nodeListToJson(SJson* pJson, const char* pName, const SNodeList*
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t jsonToNodeList(const SJson* pJson, const char* pName, SNodeList** pList) {
-  const SJson* pJsonArray = tjsonGetObjectItem(pJson, pName);
+static int32_t jsonToNodeListImpl(const SJson* pJsonArray, SNodeList** pList) {
   int32_t size = (NULL == pJsonArray ? 0 : tjsonGetArraySize(pJsonArray));
   if (size > 0) {
     *pList = nodesMakeList();
@@ -176,6 +175,10 @@ static int32_t jsonToNodeList(const SJson* pJson, const char* pName, SNodeList**
   return code;
 }
 
+static int32_t jsonToNodeList(const SJson* pJson, const char* pName, SNodeList** pList) {
+  return jsonToNodeListImpl(tjsonGetObjectItem(pJson, pName), pList);
+}
+
 static const char* jkTableMetaUid = "TableMetaUid";
 static const char* jkTableMetaSuid = "TableMetaSuid";
 
@@ -580,6 +583,8 @@ static const char* jkIntervalPhysiPlanFuncs = "Funcs";
 static const char* jkIntervalPhysiPlanInterval = "Interval";
 static const char* jkIntervalPhysiPlanOffset = "Offset";
 static const char* jkIntervalPhysiPlanSliding = "Sliding";
+static const char* jkIntervalPhysiPlanIntervalUnit = "intervalUnit";
+static const char* jkIntervalPhysiPlanSlidingUnit  = "slidingUnit";
 static const char* jkIntervalPhysiPlanFill = "Fill";
 
 static int32_t physiIntervalNodeToJson(const void* pObj, SJson* pJson) {
@@ -601,6 +606,12 @@ static int32_t physiIntervalNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddIntegerToObject(pJson, jkIntervalPhysiPlanSliding, pNode->sliding);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddIntegerToObject(pJson, jkIntervalPhysiPlanIntervalUnit, pNode->intervalUnit);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddIntegerToObject(pJson, jkIntervalPhysiPlanSlidingUnit, pNode->slidingUnit);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkIntervalPhysiPlanFill, nodeToJson, pNode->pFill);
   }
@@ -627,6 +638,12 @@ static int32_t jsonToPhysiIntervalNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetBigIntValue(pJson, jkIntervalPhysiPlanSliding, &pNode->sliding);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetTinyIntValue(pJson, jkIntervalPhysiPlanIntervalUnit, &pNode->intervalUnit);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetTinyIntValue(pJson, jkIntervalPhysiPlanSlidingUnit, &pNode->slidingUnit);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeObject(pJson, jkIntervalPhysiPlanFill, (SNode**)&pNode->pFill);
   }
@@ -1644,7 +1661,10 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
       return jsonToSubplan(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN:
       return jsonToPlan(pJson, pObj);
+    case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
+      return jsonToPhysiIntervalNode(pJson, pObj);
     default:
+      assert(0);
       break;
   }
   nodesWarn("jsonToSpecificNode unknown node = %s", nodesNodeName(nodeType(pObj)));
@@ -1710,7 +1730,7 @@ static int32_t jsonToNodeObject(const SJson* pJson, const char* pName, SNode** p
 }
 
 int32_t nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_t* pLen) {
-  if (NULL == pNode || NULL == pStr || NULL == pLen) {
+  if (NULL == pNode || NULL == pStr) {
     terrno = TSDB_CODE_FAILED;
     return TSDB_CODE_FAILED;
   }
@@ -1730,7 +1750,10 @@ int32_t nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_
   *pStr = format ? tjsonToString(pJson) : tjsonToUnformattedString(pJson);
   tjsonDelete(pJson);
 
-  *pLen = strlen(*pStr) + 1;
+  if (NULL != pLen) {
+    *pLen = strlen(*pStr) + 1;
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -1750,3 +1773,52 @@ int32_t nodesStringToNode(const char* pStr, SNode** pNode) {
   }
   return TSDB_CODE_SUCCESS;
 }
+
+int32_t nodesListToString(const SNodeList* pList, bool format, char** pStr, int32_t* pLen) {
+  if (NULL == pList || NULL == pStr || NULL == pLen) {
+    terrno = TSDB_CODE_FAILED;
+    return TSDB_CODE_FAILED;
+  }
+
+  if (0 == LIST_LENGTH(pList)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SJson* pJson = tjsonCreateArray();
+  if (NULL == pJson) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  SNode* pNode;
+  FOREACH(pNode, pList) {
+    int32_t code = tjsonAddItem(pJson, nodeToJson, pNode);
+    if (TSDB_CODE_SUCCESS != code) {
+      terrno = code;
+      return code;
+    }
+  }
+
+  *pStr = format ? tjsonToString(pJson) : tjsonToUnformattedString(pJson);
+  tjsonDelete(pJson);
+
+  *pLen = strlen(*pStr) + 1;
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t nodesStringToList(const char* pStr, SNodeList** pList) {
+  if (NULL == pStr || NULL == pList) {
+    return TSDB_CODE_SUCCESS;
+  }
+  SJson* pJson = tjsonParse(pStr);
+  if (NULL == pJson) {
+    return TSDB_CODE_FAILED;
+  }
+  int32_t code = jsonToNodeListImpl(pJson, pList);
+  if (TSDB_CODE_SUCCESS != code) {
+    nodesDestroyList(*pList);
+    terrno = code;
+    return code;
+  }
+  return TSDB_CODE_SUCCESS;
+}

source/libs/nodes/src/nodesUtilFuncs.c

@@ -76,6 +76,12 @@ SNodeptr nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SColumnDefNode));
     case QUERY_NODE_DOWNSTREAM_SOURCE:
       return makeNode(type, sizeof(SDownstreamSourceNode));
+    case QUERY_NODE_DATABASE_OPTIONS:
+      return makeNode(type, sizeof(SDatabaseOptions));
+    case QUERY_NODE_TABLE_OPTIONS:
+      return makeNode(type, sizeof(STableOptions));
+    case QUERY_NODE_INDEX_OPTIONS:
+      return makeNode(type, sizeof(SIndexOptions));
     case QUERY_NODE_SET_OPERATOR:
       return makeNode(type, sizeof(SSetOperator));
     case QUERY_NODE_SELECT_STMT:
@@ -86,6 +92,8 @@ SNodeptr nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SCreateDatabaseStmt));
     case QUERY_NODE_DROP_DATABASE_STMT:
       return makeNode(type, sizeof(SDropDatabaseStmt));
+    case QUERY_NODE_ALTER_DATABASE_STMT:
+      return makeNode(type, sizeof(SAlterDatabaseStmt));
     case QUERY_NODE_SHOW_DATABASES_STMT:
       return makeNode(type, sizeof(SShowStmt));
     case QUERY_NODE_CREATE_TABLE_STMT:
@@ -121,7 +129,20 @@ SNodeptr nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SShowStmt));
     case QUERY_NODE_SHOW_VGROUPS_STMT:
     case QUERY_NODE_SHOW_MNODES_STMT:
+    case QUERY_NODE_SHOW_QNODES_STMT:
       return makeNode(type, sizeof(SShowStmt));
+    case QUERY_NODE_CREATE_INDEX_STMT:
+      return makeNode(type, sizeof(SCreateIndexStmt));
+    case QUERY_NODE_DROP_INDEX_STMT:
+      return makeNode(type, sizeof(SDropIndexStmt));
+    case QUERY_NODE_CREATE_QNODE_STMT:
+      return makeNode(type, sizeof(SCreateQnodeStmt));
+    case QUERY_NODE_DROP_QNODE_STMT:
+      return makeNode(type, sizeof(SDropQnodeStmt));
+    case QUERY_NODE_CREATE_TOPIC_STMT:
+      return makeNode(type, sizeof(SCreateTopicStmt));
+    case QUERY_NODE_DROP_TOPIC_STMT:
+      return makeNode(type, sizeof(SDropTopicStmt));
     case QUERY_NODE_LOGIC_PLAN_SCAN:
       return makeNode(type, sizeof(SScanLogicNode));
     case QUERY_NODE_LOGIC_PLAN_JOIN:
@@ -208,6 +229,14 @@ static EDealRes destroyNode(SNode** pNode, void* pContext) {
     case QUERY_NODE_NODE_LIST:
       nodesClearList(((SNodeListNode*)(*pNode))->pNodeList);
       break;
+    case QUERY_NODE_INDEX_OPTIONS: {
+      SIndexOptions* pStmt = (SIndexOptions*)*pNode;
+      nodesDestroyList(pStmt->pFuncs);
+      nodesDestroyNode(pStmt->pInterval);
+      nodesDestroyNode(pStmt->pOffset);
+      nodesDestroyNode(pStmt->pSliding);
+      break;
+    }
     case QUERY_NODE_SELECT_STMT: {
       SSelectStmt* pStmt = (SSelectStmt*)*pNode;
       nodesDestroyList(pStmt->pProjectionList);
@@ -248,6 +277,12 @@ static EDealRes destroyNode(SNode** pNode, void* pContext) {
     case QUERY_NODE_CREATE_MULTI_TABLE_STMT:
       nodesDestroyList(((SCreateMultiTableStmt*)(*pNode))->pSubTables);
       break;
+    case QUERY_NODE_CREATE_INDEX_STMT: {
+      SCreateIndexStmt* pStmt = (SCreateIndexStmt*)*pNode;
+      nodesDestroyNode(pStmt->pOptions);
+      nodesDestroyList(pStmt->pCols);
+      break;
+    }
     default:
       break;
   }

source/libs/parser/inc/parAst.h

@@ -109,17 +109,18 @@ SNode* addLimitClause(SAstCreateContext* pCxt, SNode* pStmt, SNode* pLimit);
 SNode* createSelectStmt(SAstCreateContext* pCxt, bool isDistinct, SNodeList* pProjectionList, SNode* pTable);
 SNode* createSetOperator(SAstCreateContext* pCxt, ESetOperatorType type, SNode* pLeft, SNode* pRight);
 
-SDatabaseOptions* createDefaultDatabaseOptions(SAstCreateContext* pCxt);
-SDatabaseOptions* setDatabaseOption(SAstCreateContext* pCxt, SDatabaseOptions* pOptions, EDatabaseOptionType type, const SToken* pVal);
-SNode* createCreateDatabaseStmt(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pDbName, SDatabaseOptions* pOptions);
+SNode* createDefaultDatabaseOptions(SAstCreateContext* pCxt);
+SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOptionType type, const SToken* pVal);
+SNode* createCreateDatabaseStmt(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pDbName, SNode* pOptions);
 SNode* createDropDatabaseStmt(SAstCreateContext* pCxt, bool ignoreNotExists, const SToken* pDbName);
-STableOptions* createDefaultTableOptions(SAstCreateContext* pCxt);
-STableOptions* setTableOption(SAstCreateContext* pCxt, STableOptions* pOptions, ETableOptionType type, const SToken* pVal);
-STableOptions* setTableSmaOption(SAstCreateContext* pCxt, STableOptions* pOptions, SNodeList* pSma);
+SNode* createAlterDatabaseStmt(SAstCreateContext* pCxt, const SToken* pDbName, SNode* pOptions);
+SNode* createDefaultTableOptions(SAstCreateContext* pCxt);
+SNode* setTableOption(SAstCreateContext* pCxt, SNode* pOptions, ETableOptionType type, const SToken* pVal);
+SNode* setTableSmaOption(SAstCreateContext* pCxt, SNode* pOptions, SNodeList* pSma);
 SNode* createColumnDefNode(SAstCreateContext* pCxt, const SToken* pColName, SDataType dataType, const SToken* pComment);
 SDataType createDataType(uint8_t type);
 SDataType createVarLenDataType(uint8_t type, const SToken* pLen);
-SNode* createCreateTableStmt(SAstCreateContext* pCxt, bool ignoreExists, SNode* pRealTable, SNodeList* pCols, SNodeList* pTags, STableOptions* pOptions);
+SNode* createCreateTableStmt(SAstCreateContext* pCxt, bool ignoreExists, SNode* pRealTable, SNodeList* pCols, SNodeList* pTags, SNode* pOptions);
 SNode* createCreateSubTableClause(SAstCreateContext* pCxt, bool ignoreExists, SNode* pRealTable, SNode* pUseRealTable, SNodeList* pSpecificTags, SNodeList* pValsOfTags);
 SNode* createCreateMultiTableStmt(SAstCreateContext* pCxt, SNodeList* pSubTables);
 SNode* createDropTableClause(SAstCreateContext* pCxt, bool ignoreNotExists, SNode* pRealTable);
@@ -132,6 +133,13 @@ SNode* createAlterUserStmt(SAstCreateContext* pCxt, const SToken* pUserName, int
 SNode* createDropUserStmt(SAstCreateContext* pCxt, const SToken* pUserName);
 SNode* createCreateDnodeStmt(SAstCreateContext* pCxt, const SToken* pFqdn, const SToken* pPort);
 SNode* createDropDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode);
+SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, const SToken* pIndexName, const SToken* pTableName, SNodeList* pCols, SNode* pOptions);
+SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInterval, SNode* pOffset, SNode* pSliding);
+SNode* createDropIndexStmt(SAstCreateContext* pCxt, const SToken* pIndexName, const SToken* pTableName);
+SNode* createCreateQnodeStmt(SAstCreateContext* pCxt, const SToken* pDnodeId);
+SNode* createDropQnodeStmt(SAstCreateContext* pCxt, const SToken* pDnodeId);
+SNode* createCreateTopicStmt(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pTopicName, SNode* pQuery, const SToken* pSubscribeDbName);
+SNode* createDropTopicStmt(SAstCreateContext* pCxt, bool ignoreNotExists, const SToken* pTopicName);
 
 #ifdef __cplusplus
 }

source/libs/parser/src/parInsert.c

@@ -763,6 +763,8 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pTagsSchema,
   // todo construct payload
 
   tfree(row);
+
+  return 0;
 }
 
 // pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)

source/libs/parser/src/parTokenizer.c

@@ -59,11 +59,13 @@ static SKeyword keywordTable[] = {
     {"FLOAT",         TK_FLOAT},
     {"FROM",          TK_FROM},
     {"FSYNC",         TK_FSYNC},
+    {"FUNCTION",      TK_FUNCTION},
     {"GROUP",         TK_GROUP},
     {"HAVING",        TK_HAVING},
     {"IF",            TK_IF},
     {"IMPORT",        TK_IMPORT},
     {"IN",            TK_IN},
+    {"INDEX",         TK_INDEX},
     {"INNER",         TK_INNER},
     {"INT",           TK_INT},
     {"INSERT",        TK_INSERT},
@@ -97,6 +99,8 @@ static SKeyword keywordTable[] = {
     {"PRECISION",     TK_PRECISION},
     {"PRIVILEGE",     TK_PRIVILEGE},
     {"PREV",          TK_PREV},
+    {"QNODE",         TK_QNODE},
+    {"QNODES",        TK_QNODES},
     {"QUORUM",        TK_QUORUM},
     {"REPLICA",       TK_REPLICA},
     {"SELECT",        TK_SELECT},
@@ -117,6 +121,7 @@ static SKeyword keywordTable[] = {
     {"TAGS",          TK_TAGS},
     {"TIMESTAMP",     TK_TIMESTAMP},
     {"TINYINT",       TK_TINYINT},
+    {"TOPIC",         TK_TOPIC},
     {"TTL",           TK_TTL},
     {"UNION",         TK_UNION},
     {"UNSIGNED",      TK_UNSIGNED},
@@ -226,11 +231,9 @@ static SKeyword keywordTable[] = {
     // {"TBNAME",       TK_TBNAME},
     // {"VNODES",       TK_VNODES},
 //    {"PARTITIONS",   TK_PARTITIONS},
-    // {"TOPIC",        TK_TOPIC},
     // {"TOPICS",       TK_TOPICS},
     // {"COMPACT",      TK_COMPACT},
     // {"MODIFY",       TK_MODIFY},
-    // {"FUNCTION",     TK_FUNCTION},
     // {"FUNCTIONS",    TK_FUNCTIONS},
     // {"OUTPUTTYPE",   TK_OUTPUTTYPE},
     // {"AGGREGATE",    TK_AGGREGATE},

source/libs/planner/src/planPhysiCreater.c

@@ -480,6 +480,9 @@ static SPhysiNode* createIntervalPhysiNode(SPhysiPlanContext* pCxt, SNodeList* p
   pInterval->interval = pWindowLogicNode->interval;
   pInterval->offset = pWindowLogicNode->offset;
   pInterval->sliding = pWindowLogicNode->sliding;
+  pInterval->intervalUnit = pWindowLogicNode->intervalUnit;
+  pInterval->slidingUnit = pWindowLogicNode->slidingUnit;
+
   pInterval->pFill = nodesCloneNode(pWindowLogicNode->pFill);
 
   SNodeList* pPrecalcExprs = NULL;

source/libs/scalar/src/scalar.c

@@ -239,6 +239,9 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
       
       break;
     }
+
+    default:
+      break;
   }
 
   if (param->num > *rowNum) {