Merge branch '3.0' into feature/3.0_liaohj

cmake/cmake.define

@@ -14,6 +14,25 @@ MESSAGE(STATUS "Project binary files output path: " ${PROJECT_BINARY_DIR})
 MESSAGE(STATUS "Project executable files output path: " ${EXECUTABLE_OUTPUT_PATH})
 MESSAGE(STATUS "Project library files output path: " ${LIBRARY_OUTPUT_PATH})
 
+find_package(Git QUIET)
+if(GIT_FOUND AND EXISTS "${PROJECT_SOURCE_DIR}/.git")
+# Update submodules as needed
+    option(GIT_SUBMODULE "Check submodules during build" ON)
+    if(GIT_SUBMODULE)
+        message(STATUS "Submodule update")
+        execute_process(COMMAND ${GIT_EXECUTABLE} submodule update --init --recursive
+                        WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
+                        RESULT_VARIABLE GIT_SUBMOD_RESULT)
+        if(NOT GIT_SUBMOD_RESULT EQUAL "0")
+            message(WARNING "git submodule update --init --recursive failed with ${GIT_SUBMOD_RESULT}, please checkout submodules")
+        endif()
+    endif()
+endif()
+
+if(NOT EXISTS "${PROJECT_SOURCE_DIR}/tools/taos-tools/CMakeLists.txt")
+    message(WARNING "The submodules were not downloaded! GIT_SUBMODULE was turned off or failed. Please update submodules manually if you need build them.")
+endif()
+
 if (NOT DEFINED TD_GRANT)
   SET(TD_GRANT FALSE)
 endif()

example/src/tmq.c

@@ -101,8 +101,8 @@ int32_t create_topic() {
   }
   taos_free_result(pRes);
 
-  pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");
-  /*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");*/
+  /*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
+  pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
     return -1;

include/common/tcommon.h

@@ -45,6 +45,12 @@ enum {
   STREAM_TRIGGER__BY_EVENT_TIME,
 };
 
+typedef enum EStreamType {
+  STREAM_NORMAL = 1,
+  STREAM_INVERT,
+  STREAM_INVALID,
+} EStreamType;
+
 typedef struct {
   uint32_t  numOfTables;
   SArray*   pGroupList;
@@ -71,6 +77,7 @@ typedef struct SDataBlockInfo {
   int16_t     numOfCols;
   int16_t     hasVarCol;
   int32_t     capacity;
+  EStreamType type;
 } SDataBlockInfo;
 
 typedef struct SSDataBlock {

include/common/tdatablock.h

@@ -56,11 +56,11 @@ SEpSet getEpSet_s(SCorEpSet* pEpSet);
 
 #define colDataSetNotNull_f(bm_, r_)                  \
   do {                                                \
-    BMCharPos(bm_, r_) &= ~(1u << (7u - BitPos(r_)));  \
+    BMCharPos(bm_, r_) &= ~(1u << (7u - BitPos(r_))); \
   } while (0)
 
-#define colDataIsNull_var(pColumnInfoData, row) (pColumnInfoData->varmeta.offset[row] == -1)
-#define colDataSetNull_var(pColumnInfoData, row)  (pColumnInfoData->varmeta.offset[row] = -1)
+#define colDataIsNull_var(pColumnInfoData, row)  (pColumnInfoData->varmeta.offset[row] == -1)
+#define colDataSetNull_var(pColumnInfoData, row) (pColumnInfoData->varmeta.offset[row] = -1)
 
 #define BitmapLen(_n) (((_n) + ((1 << NBIT) - 1)) >> NBIT)
 
@@ -187,8 +187,8 @@ static FORCE_INLINE void colDataAppendDouble(SColumnInfoData* pColumnInfoData, u
 }
 
 int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull);
-int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, int32_t* capacity, const SColumnInfoData* pSource,
-                        uint32_t numOfRow2);
+int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, int32_t* capacity,
+                        const SColumnInfoData* pSource, uint32_t numOfRow2);
 int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows);
 int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock);
 
@@ -230,9 +230,9 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData);
 void blockDebugShowData(const SArray* dataBlocks);
 
 int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks, STSchema* pTSchema, int32_t vgId,
-                                 tb_uid_t uid, tb_uid_t suid);
+                                    tb_uid_t uid, tb_uid_t suid);
 
-SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema);
+SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool createTb, int64_t suid);
 
 static FORCE_INLINE int32_t blockGetEncodeSize(const SSDataBlock* pBlock) {
   return blockDataGetSerialMetaSize(pBlock) + blockDataGetSize(pBlock);

include/common/tmsg.h

@@ -135,6 +135,8 @@ typedef enum _mgmt_table {
 #define TSDB_ALTER_USER_ADD_WRITE_DB    0x6
 #define TSDB_ALTER_USER_REMOVE_WRITE_DB 0x7
 #define TSDB_ALTER_USER_CLEAR_WRITE_DB  0x8
+#define TSDB_ALTER_USER_ADD_ALL_DB      0x9
+#define TSDB_ALTER_USER_REMOVE_ALL_DB   0xA
 
 #define TSDB_ALTER_USER_PRIVILEGES 0x2
 
@@ -338,8 +340,8 @@ typedef struct SEpSet {
   SEp    eps[TSDB_MAX_REPLICA];
 } SEpSet;
 
-int32_t tEncodeSEpSet(SCoder* pEncoder, const SEpSet* pEp);
-int32_t tDecodeSEpSet(SCoder* pDecoder, SEpSet* pEp);
+int32_t tEncodeSEpSet(SEncoder* pEncoder, const SEpSet* pEp);
+int32_t tDecodeSEpSet(SDecoder* pDecoder, SEpSet* pEp);
 int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp);
 void*   taosDecodeSEpSet(const void* buf, SEpSet* pEp);
 
@@ -611,8 +613,8 @@ typedef struct {
 
 int32_t tSerializeSUseDbRsp(void* buf, int32_t bufLen, const SUseDbRsp* pRsp);
 int32_t tDeserializeSUseDbRsp(void* buf, int32_t bufLen, SUseDbRsp* pRsp);
-int32_t tSerializeSUseDbRspImp(SCoder* pEncoder, const SUseDbRsp* pRsp);
-int32_t tDeserializeSUseDbRspImp(SCoder* pDecoder, SUseDbRsp* pRsp);
+int32_t tSerializeSUseDbRspImp(SEncoder* pEncoder, const SUseDbRsp* pRsp);
+int32_t tDeserializeSUseDbRspImp(SDecoder* pDecoder, SUseDbRsp* pRsp);
 void    tFreeSUsedbRsp(SUseDbRsp* pRsp);
 
 typedef struct {
@@ -1527,8 +1529,8 @@ typedef struct {
   char*   qmsg2;  // pAst2:qmsg2:SRetention2 => trigger aggr task2
 } SRSmaParam;
 
-int32_t tEncodeSRSmaParam(SCoder* pCoder, const SRSmaParam* pRSmaParam);
-int32_t tDecodeSRSmaParam(SCoder* pCoder, SRSmaParam* pRSmaParam);
+int32_t tEncodeSRSmaParam(SEncoder* pCoder, const SRSmaParam* pRSmaParam);
+int32_t tDecodeSRSmaParam(SDecoder* pCoder, SRSmaParam* pRSmaParam);
 
 // TDMT_VND_CREATE_STB ==============
 typedef struct SVCreateStbReq {
@@ -1540,8 +1542,8 @@ typedef struct SVCreateStbReq {
   SRSmaParam     pRSmaParam;
 } SVCreateStbReq;
 
-int tEncodeSVCreateStbReq(SCoder* pCoder, const SVCreateStbReq* pReq);
-int tDecodeSVCreateStbReq(SCoder* pCoder, SVCreateStbReq* pReq);
+int tEncodeSVCreateStbReq(SEncoder* pCoder, const SVCreateStbReq* pReq);
+int tDecodeSVCreateStbReq(SDecoder* pCoder, SVCreateStbReq* pReq);
 
 // TDMT_VND_DROP_STB ==============
 typedef struct SVDropStbReq {
@@ -1549,8 +1551,8 @@ typedef struct SVDropStbReq {
   tb_uid_t    suid;
 } SVDropStbReq;
 
-int32_t tEncodeSVDropStbReq(SCoder* pCoder, const SVDropStbReq* pReq);
-int32_t tDecodeSVDropStbReq(SCoder* pCoder, SVDropStbReq* pReq);
+int32_t tEncodeSVDropStbReq(SEncoder* pCoder, const SVDropStbReq* pReq);
+int32_t tDecodeSVDropStbReq(SDecoder* pCoder, SVDropStbReq* pReq);
 
 #define TD_CREATE_IF_NOT_EXISTS 0x1
 typedef struct SVCreateTbReq {
@@ -1562,8 +1564,8 @@ typedef struct SVCreateTbReq {
   int8_t      type;
   union {
     struct {
-      tb_uid_t    suid;
-      const void* pTag;
+      tb_uid_t       suid;
+      const uint8_t* pTag;
     } ctb;
     struct {
       SSchemaWrapper schema;
@@ -1571,8 +1573,8 @@ typedef struct SVCreateTbReq {
   };
 } SVCreateTbReq;
 
-int tEncodeSVCreateTbReq(SCoder* pCoder, const SVCreateTbReq* pReq);
-int tDecodeSVCreateTbReq(SCoder* pCoder, SVCreateTbReq* pReq);
+int tEncodeSVCreateTbReq(SEncoder* pCoder, const SVCreateTbReq* pReq);
+int tDecodeSVCreateTbReq(SDecoder* pCoder, SVCreateTbReq* pReq);
 
 typedef struct {
   int32_t nReqs;
@@ -1582,15 +1584,15 @@ typedef struct {
   };
 } SVCreateTbBatchReq;
 
-int tEncodeSVCreateTbBatchReq(SCoder* pCoder, const SVCreateTbBatchReq* pReq);
-int tDecodeSVCreateTbBatchReq(SCoder* pCoder, SVCreateTbBatchReq* pReq);
+int tEncodeSVCreateTbBatchReq(SEncoder* pCoder, const SVCreateTbBatchReq* pReq);
+int tDecodeSVCreateTbBatchReq(SDecoder* pCoder, SVCreateTbBatchReq* pReq);
 
 typedef struct {
   int32_t code;
 } SVCreateTbRsp, SVUpdateTbRsp;
 
-int tEncodeSVCreateTbRsp(SCoder* pCoder, const SVCreateTbRsp* pRsp);
-int tDecodeSVCreateTbRsp(SCoder* pCoder, SVCreateTbRsp* pRsp);
+int tEncodeSVCreateTbRsp(SEncoder* pCoder, const SVCreateTbRsp* pRsp);
+int tDecodeSVCreateTbRsp(SDecoder* pCoder, SVCreateTbRsp* pRsp);
 
 int32_t tSerializeSVCreateTbReq(void** buf, SVCreateTbReq* pReq);
 void*   tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);
@@ -1603,8 +1605,8 @@ typedef struct {
   };
 } SVCreateTbBatchRsp;
 
-int tEncodeSVCreateTbBatchRsp(SCoder* pCoder, const SVCreateTbBatchRsp* pRsp);
-int tDecodeSVCreateTbBatchRsp(SCoder* pCoder, SVCreateTbBatchRsp* pRsp);
+int tEncodeSVCreateTbBatchRsp(SEncoder* pCoder, const SVCreateTbBatchRsp* pRsp);
+int tDecodeSVCreateTbBatchRsp(SDecoder* pCoder, SVCreateTbBatchRsp* pRsp);
 
 int32_t tSerializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatchRsp* pRsp);
 int32_t tDeserializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatchRsp* pRsp);
@@ -1627,8 +1629,8 @@ typedef struct {
   };
 } SVDropTbBatchReq;
 
-int32_t tEncodeSVDropTbBatchReq(SCoder* pCoder, const SVDropTbBatchReq* pReq);
-int32_t tDecodeSVDropTbBatchReq(SCoder* pCoder, SVDropTbBatchReq* pReq);
+int32_t tEncodeSVDropTbBatchReq(SEncoder* pCoder, const SVDropTbBatchReq* pReq);
+int32_t tDecodeSVDropTbBatchReq(SDecoder* pCoder, SVDropTbBatchReq* pReq);
 
 typedef struct {
   int32_t nRsps;
@@ -1638,8 +1640,8 @@ typedef struct {
   };
 } SVDropTbBatchRsp;
 
-int32_t tEncodeSVDropTbBatchRsp(SCoder* pCoder, const SVDropTbBatchRsp* pRsp);
-int32_t tDecodeSVDropTbBatchRsp(SCoder* pCoder, SVDropTbBatchRsp* pRsp);
+int32_t tEncodeSVDropTbBatchRsp(SEncoder* pCoder, const SVDropTbBatchRsp* pRsp);
+int32_t tDecodeSVDropTbBatchRsp(SDecoder* pCoder, SVDropTbBatchRsp* pRsp);
 
 typedef struct {
   SMsgHead head;
@@ -1821,14 +1823,14 @@ static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) {
 int32_t tSerializeSClientHbBatchRsp(void* buf, int32_t bufLen, const SClientHbBatchRsp* pBatchRsp);
 int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchRsp* pBatchRsp);
 
-static FORCE_INLINE int32_t tEncodeSKv(SCoder* pEncoder, const SKv* pKv) {
+static FORCE_INLINE int32_t tEncodeSKv(SEncoder* pEncoder, const SKv* pKv) {
   if (tEncodeI32(pEncoder, pKv->key) < 0) return -1;
   if (tEncodeI32(pEncoder, pKv->valueLen) < 0) return -1;
-  if (tEncodeBinary(pEncoder, (const char*)pKv->value, pKv->valueLen) < 0) return -1;
+  if (tEncodeBinary(pEncoder, (uint8_t*)pKv->value, pKv->valueLen) < 0) return -1;
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) {
+static FORCE_INLINE int32_t tDecodeSKv(SDecoder* pDecoder, SKv* pKv) {
   if (tDecodeI32(pDecoder, &pKv->key) < 0) return -1;
   if (tDecodeI32(pDecoder, &pKv->valueLen) < 0) return -1;
   pKv->value = taosMemoryMalloc(pKv->valueLen + 1);
@@ -1837,13 +1839,13 @@ static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) {
   return 0;
 }
 
-static FORCE_INLINE int32_t tEncodeSClientHbKey(SCoder* pEncoder, const SClientHbKey* pKey) {
+static FORCE_INLINE int32_t tEncodeSClientHbKey(SEncoder* pEncoder, const SClientHbKey* pKey) {
   if (tEncodeI64(pEncoder, pKey->tscRid) < 0) return -1;
   if (tEncodeI8(pEncoder, pKey->connType) < 0) return -1;
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeSClientHbKey(SCoder* pDecoder, SClientHbKey* pKey) {
+static FORCE_INLINE int32_t tDecodeSClientHbKey(SDecoder* pDecoder, SClientHbKey* pKey) {
   if (tDecodeI64(pDecoder, &pKey->tscRid) < 0) return -1;
   if (tDecodeI8(pDecoder, &pKey->connType) < 0) return -1;
   return 0;
@@ -2046,10 +2048,10 @@ typedef struct {
   int32_t reserved;
 } SMqCMCommitOffsetRsp;
 
-int32_t tEncodeSMqOffset(SCoder* encoder, const SMqOffset* pOffset);
-int32_t tDecodeSMqOffset(SCoder* decoder, SMqOffset* pOffset);
-int32_t tEncodeSMqCMCommitOffsetReq(SCoder* encoder, const SMqCMCommitOffsetReq* pReq);
-int32_t tDecodeSMqCMCommitOffsetReq(SCoder* decoder, SMqCMCommitOffsetReq* pReq);
+int32_t tEncodeSMqOffset(SEncoder* encoder, const SMqOffset* pOffset);
+int32_t tDecodeSMqOffset(SDecoder* decoder, SMqOffset* pOffset);
+int32_t tEncodeSMqCMCommitOffsetReq(SEncoder* encoder, const SMqCMCommitOffsetReq* pReq);
+int32_t tDecodeSMqCMCommitOffsetReq(SDecoder* decoder, SMqCMCommitOffsetReq* pReq);
 
 static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* pSchemaWrapper) {
   SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
@@ -2089,7 +2091,7 @@ static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) {
   return (void*)buf;
 }
 
-static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) {
+static FORCE_INLINE int32_t tEncodeSSchema(SEncoder* pEncoder, const SSchema* pSchema) {
   if (tEncodeI8(pEncoder, pSchema->type) < 0) return -1;
   if (tEncodeI8(pEncoder, pSchema->flags) < 0) return -1;
   if (tEncodeI32v(pEncoder, pSchema->bytes) < 0) return -1;
@@ -2098,7 +2100,7 @@ static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSch
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeSSchema(SCoder* pDecoder, SSchema* pSchema) {
+static FORCE_INLINE int32_t tDecodeSSchema(SDecoder* pDecoder, SSchema* pSchema) {
   if (tDecodeI8(pDecoder, &pSchema->type) < 0) return -1;
   if (tDecodeI8(pDecoder, &pSchema->flags) < 0) return -1;
   if (tDecodeI32v(pDecoder, &pSchema->bytes) < 0) return -1;
@@ -2131,7 +2133,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapp
   return (void*)buf;
 }
 
-static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchemaWrapper* pSW) {
+static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SEncoder* pEncoder, const SSchemaWrapper* pSW) {
   if (tEncodeI32v(pEncoder, pSW->nCols) < 0) return -1;
   if (tEncodeI32v(pEncoder, pSW->sver) < 0) return -1;
   for (int32_t i = 0; i < pSW->nCols; i++) {
@@ -2141,7 +2143,7 @@ static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchem
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SCoder* pDecoder, SSchemaWrapper* pSW) {
+static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SDecoder* pDecoder, SSchemaWrapper* pSW) {
   if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1;
   if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1;
 
@@ -2593,12 +2595,12 @@ static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) {
 
 #define TD_AUTO_CREATE_TABLE 0x1
 typedef struct {
-  int64_t       suid;
-  int64_t       uid;
-  int32_t       sver;
-  uint64_t      nData;
-  const void*   pData;
-  SVCreateTbReq cTbReq;
+  int64_t        suid;
+  int64_t        uid;
+  int32_t        sver;
+  uint32_t       nData;
+  const uint8_t* pData;
+  SVCreateTbReq  cTbReq;
 } SVSubmitBlk;
 
 typedef struct {
@@ -2610,8 +2612,8 @@ typedef struct {
   };
 } SVSubmitReq;
 
-int32_t tEncodeSVSubmitReq(SCoder* pCoder, const SVSubmitReq* pReq);
-int32_t tDecodeSVSubmitReq(SCoder* pCoder, SVSubmitReq* pReq);
+int32_t tEncodeSVSubmitReq(SEncoder* pCoder, const SVSubmitReq* pReq);
+int32_t tDecodeSVSubmitReq(SDecoder* pCoder, SVSubmitReq* pReq);
 
 #pragma pack(pop)
 

include/common/ttokendef.h

@@ -51,201 +51,208 @@
 #define TK_USER          33
 #define TK_PRIVILEGE     34
 #define TK_DROP          35
-#define TK_DNODE         36
-#define TK_PORT          37
-#define TK_NK_INTEGER    38
-#define TK_DNODES        39
-#define TK_NK_IPTOKEN    40
-#define TK_LOCAL         41
-#define TK_QNODE         42
-#define TK_ON            43
-#define TK_BNODE         44
-#define TK_SNODE         45
-#define TK_MNODE         46
-#define TK_DATABASE      47
-#define TK_USE           48
-#define TK_IF            49
-#define TK_NOT           50
-#define TK_EXISTS        51
-#define TK_BUFFER        52
-#define TK_CACHELAST     53
-#define TK_COMP          54
-#define TK_DAYS          55
-#define TK_NK_VARIABLE   56
-#define TK_FSYNC         57
-#define TK_MAXROWS       58
-#define TK_MINROWS       59
-#define TK_KEEP          60
-#define TK_PAGES         61
-#define TK_PAGESIZE      62
-#define TK_PRECISION     63
-#define TK_REPLICA       64
-#define TK_STRICT        65
-#define TK_WAL           66
-#define TK_VGROUPS       67
-#define TK_SINGLE_STABLE 68
-#define TK_RETENTIONS    69
-#define TK_NK_COMMA      70
-#define TK_NK_COLON      71
-#define TK_TABLE         72
-#define TK_NK_LP         73
-#define TK_NK_RP         74
-#define TK_STABLE        75
-#define TK_ADD           76
-#define TK_COLUMN        77
-#define TK_MODIFY        78
-#define TK_RENAME        79
-#define TK_TAG           80
-#define TK_SET           81
-#define TK_NK_EQ         82
-#define TK_USING         83
-#define TK_TAGS          84
-#define TK_NK_DOT        85
-#define TK_COMMENT       86
-#define TK_BOOL          87
-#define TK_TINYINT       88
-#define TK_SMALLINT      89
-#define TK_INT           90
-#define TK_INTEGER       91
-#define TK_BIGINT        92
-#define TK_FLOAT         93
-#define TK_DOUBLE        94
-#define TK_BINARY        95
-#define TK_TIMESTAMP     96
-#define TK_NCHAR         97
-#define TK_UNSIGNED      98
-#define TK_JSON          99
-#define TK_VARCHAR       100
-#define TK_MEDIUMBLOB    101
-#define TK_BLOB          102
-#define TK_VARBINARY     103
-#define TK_DECIMAL       104
-#define TK_DELAY         105
-#define TK_FILE_FACTOR   106
-#define TK_NK_FLOAT      107
-#define TK_ROLLUP        108
-#define TK_TTL           109
-#define TK_SMA           110
-#define TK_SHOW          111
-#define TK_DATABASES     112
-#define TK_TABLES        113
-#define TK_STABLES       114
-#define TK_MNODES        115
-#define TK_MODULES       116
-#define TK_QNODES        117
-#define TK_FUNCTIONS     118
-#define TK_INDEXES       119
-#define TK_FROM          120
-#define TK_ACCOUNTS      121
-#define TK_APPS          122
-#define TK_CONNECTIONS   123
-#define TK_LICENCE       124
-#define TK_GRANTS        125
-#define TK_QUERIES       126
-#define TK_SCORES        127
-#define TK_TOPICS        128
-#define TK_VARIABLES     129
-#define TK_BNODES        130
-#define TK_SNODES        131
-#define TK_CLUSTER       132
-#define TK_LIKE          133
-#define TK_INDEX         134
-#define TK_FULLTEXT      135
-#define TK_FUNCTION      136
-#define TK_INTERVAL      137
-#define TK_TOPIC         138
-#define TK_AS            139
-#define TK_WITH          140
-#define TK_SCHEMA        141
-#define TK_DESC          142
-#define TK_DESCRIBE      143
-#define TK_RESET         144
-#define TK_QUERY         145
-#define TK_CACHE         146
-#define TK_EXPLAIN       147
-#define TK_ANALYZE       148
-#define TK_VERBOSE       149
-#define TK_NK_BOOL       150
-#define TK_RATIO         151
-#define TK_COMPACT       152
-#define TK_VNODES        153
-#define TK_IN            154
-#define TK_OUTPUTTYPE    155
-#define TK_AGGREGATE     156
-#define TK_BUFSIZE       157
-#define TK_STREAM        158
-#define TK_INTO          159
-#define TK_TRIGGER       160
-#define TK_AT_ONCE       161
-#define TK_WINDOW_CLOSE  162
-#define TK_WATERMARK     163
-#define TK_KILL          164
-#define TK_CONNECTION    165
-#define TK_MERGE         166
-#define TK_VGROUP        167
-#define TK_REDISTRIBUTE  168
-#define TK_SPLIT         169
-#define TK_SYNCDB        170
-#define TK_NULL          171
-#define TK_NK_QUESTION   172
-#define TK_NK_ARROW      173
-#define TK_ROWTS         174
-#define TK_TBNAME        175
-#define TK_QSTARTTS      176
-#define TK_QENDTS        177
-#define TK_WSTARTTS      178
-#define TK_WENDTS        179
-#define TK_WDURATION     180
-#define TK_CAST          181
-#define TK_NOW           182
-#define TK_TODAY         183
-#define TK_TIMEZONE      184
-#define TK_COUNT         185
-#define TK_FIRST         186
-#define TK_LAST          187
-#define TK_LAST_ROW      188
-#define TK_BETWEEN       189
-#define TK_IS            190
-#define TK_NK_LT         191
-#define TK_NK_GT         192
-#define TK_NK_LE         193
-#define TK_NK_GE         194
-#define TK_NK_NE         195
-#define TK_MATCH         196
-#define TK_NMATCH        197
-#define TK_CONTAINS      198
-#define TK_JOIN          199
-#define TK_INNER         200
-#define TK_SELECT        201
-#define TK_DISTINCT      202
-#define TK_WHERE         203
-#define TK_PARTITION     204
-#define TK_BY            205
-#define TK_SESSION       206
-#define TK_STATE_WINDOW  207
-#define TK_SLIDING       208
-#define TK_FILL          209
-#define TK_VALUE         210
-#define TK_NONE          211
-#define TK_PREV          212
-#define TK_LINEAR        213
-#define TK_NEXT          214
-#define TK_GROUP         215
-#define TK_HAVING        216
-#define TK_ORDER         217
-#define TK_SLIMIT        218
-#define TK_SOFFSET       219
-#define TK_LIMIT         220
-#define TK_OFFSET        221
-#define TK_ASC           222
-#define TK_NULLS         223
-#define TK_ID            224
-#define TK_NK_BITNOT     225
-#define TK_INSERT        226
-#define TK_VALUES        227
-#define TK_IMPORT        228
-#define TK_NK_SEMI       229
-#define TK_FILE          230
+#define TK_GRANT         36
+#define TK_ON            37
+#define TK_TO            38
+#define TK_REVOKE        39
+#define TK_FROM          40
+#define TK_NK_COMMA      41
+#define TK_READ          42
+#define TK_WRITE         43
+#define TK_NK_DOT        44
+#define TK_DNODE         45
+#define TK_PORT          46
+#define TK_NK_INTEGER    47
+#define TK_DNODES        48
+#define TK_NK_IPTOKEN    49
+#define TK_LOCAL         50
+#define TK_QNODE         51
+#define TK_BNODE         52
+#define TK_SNODE         53
+#define TK_MNODE         54
+#define TK_DATABASE      55
+#define TK_USE           56
+#define TK_IF            57
+#define TK_NOT           58
+#define TK_EXISTS        59
+#define TK_BUFFER        60
+#define TK_CACHELAST     61
+#define TK_COMP          62
+#define TK_DAYS          63
+#define TK_NK_VARIABLE   64
+#define TK_FSYNC         65
+#define TK_MAXROWS       66
+#define TK_MINROWS       67
+#define TK_KEEP          68
+#define TK_PAGES         69
+#define TK_PAGESIZE      70
+#define TK_PRECISION     71
+#define TK_REPLICA       72
+#define TK_STRICT        73
+#define TK_WAL           74
+#define TK_VGROUPS       75
+#define TK_SINGLE_STABLE 76
+#define TK_RETENTIONS    77
+#define TK_NK_COLON      78
+#define TK_TABLE         79
+#define TK_NK_LP         80
+#define TK_NK_RP         81
+#define TK_STABLE        82
+#define TK_ADD           83
+#define TK_COLUMN        84
+#define TK_MODIFY        85
+#define TK_RENAME        86
+#define TK_TAG           87
+#define TK_SET           88
+#define TK_NK_EQ         89
+#define TK_USING         90
+#define TK_TAGS          91
+#define TK_COMMENT       92
+#define TK_BOOL          93
+#define TK_TINYINT       94
+#define TK_SMALLINT      95
+#define TK_INT           96
+#define TK_INTEGER       97
+#define TK_BIGINT        98
+#define TK_FLOAT         99
+#define TK_DOUBLE        100
+#define TK_BINARY        101
+#define TK_TIMESTAMP     102
+#define TK_NCHAR         103
+#define TK_UNSIGNED      104
+#define TK_JSON          105
+#define TK_VARCHAR       106
+#define TK_MEDIUMBLOB    107
+#define TK_BLOB          108
+#define TK_VARBINARY     109
+#define TK_DECIMAL       110
+#define TK_DELAY         111
+#define TK_FILE_FACTOR   112
+#define TK_NK_FLOAT      113
+#define TK_ROLLUP        114
+#define TK_TTL           115
+#define TK_SMA           116
+#define TK_SHOW          117
+#define TK_DATABASES     118
+#define TK_TABLES        119
+#define TK_STABLES       120
+#define TK_MNODES        121
+#define TK_MODULES       122
+#define TK_QNODES        123
+#define TK_FUNCTIONS     124
+#define TK_INDEXES       125
+#define TK_ACCOUNTS      126
+#define TK_APPS          127
+#define TK_CONNECTIONS   128
+#define TK_LICENCE       129
+#define TK_GRANTS        130
+#define TK_QUERIES       131
+#define TK_SCORES        132
+#define TK_TOPICS        133
+#define TK_VARIABLES     134
+#define TK_BNODES        135
+#define TK_SNODES        136
+#define TK_CLUSTER       137
+#define TK_TRANSACTIONS  138
+#define TK_LIKE          139
+#define TK_INDEX         140
+#define TK_FULLTEXT      141
+#define TK_FUNCTION      142
+#define TK_INTERVAL      143
+#define TK_TOPIC         144
+#define TK_AS            145
+#define TK_WITH          146
+#define TK_SCHEMA        147
+#define TK_DESC          148
+#define TK_DESCRIBE      149
+#define TK_RESET         150
+#define TK_QUERY         151
+#define TK_CACHE         152
+#define TK_EXPLAIN       153
+#define TK_ANALYZE       154
+#define TK_VERBOSE       155
+#define TK_NK_BOOL       156
+#define TK_RATIO         157
+#define TK_COMPACT       158
+#define TK_VNODES        159
+#define TK_IN            160
+#define TK_OUTPUTTYPE    161
+#define TK_AGGREGATE     162
+#define TK_BUFSIZE       163
+#define TK_STREAM        164
+#define TK_INTO          165
+#define TK_TRIGGER       166
+#define TK_AT_ONCE       167
+#define TK_WINDOW_CLOSE  168
+#define TK_WATERMARK     169
+#define TK_KILL          170
+#define TK_CONNECTION    171
+#define TK_TRANSACTION   172
+#define TK_MERGE         173
+#define TK_VGROUP        174
+#define TK_REDISTRIBUTE  175
+#define TK_SPLIT         176
+#define TK_SYNCDB        177
+#define TK_NULL          178
+#define TK_NK_QUESTION   179
+#define TK_NK_ARROW      180
+#define TK_ROWTS         181
+#define TK_TBNAME        182
+#define TK_QSTARTTS      183
+#define TK_QENDTS        184
+#define TK_WSTARTTS      185
+#define TK_WENDTS        186
+#define TK_WDURATION     187
+#define TK_CAST          188
+#define TK_NOW           189
+#define TK_TODAY         190
+#define TK_TIMEZONE      191
+#define TK_COUNT         192
+#define TK_FIRST         193
+#define TK_LAST          194
+#define TK_LAST_ROW      195
+#define TK_BETWEEN       196
+#define TK_IS            197
+#define TK_NK_LT         198
+#define TK_NK_GT         199
+#define TK_NK_LE         200
+#define TK_NK_GE         201
+#define TK_NK_NE         202
+#define TK_MATCH         203
+#define TK_NMATCH        204
+#define TK_CONTAINS      205
+#define TK_JOIN          206
+#define TK_INNER         207
+#define TK_SELECT        208
+#define TK_DISTINCT      209
+#define TK_WHERE         210
+#define TK_PARTITION     211
+#define TK_BY            212
+#define TK_SESSION       213
+#define TK_STATE_WINDOW  214
+#define TK_SLIDING       215
+#define TK_FILL          216
+#define TK_VALUE         217
+#define TK_NONE          218
+#define TK_PREV          219
+#define TK_LINEAR        220
+#define TK_NEXT          221
+#define TK_GROUP         222
+#define TK_HAVING        223
+#define TK_ORDER         224
+#define TK_SLIMIT        225
+#define TK_SOFFSET       226
+#define TK_LIMIT         227
+#define TK_OFFSET        228
+#define TK_ASC           229
+#define TK_NULLS         230
+#define TK_ID            231
+#define TK_NK_BITNOT     232
+#define TK_INSERT        233
+#define TK_VALUES        234
+#define TK_IMPORT        235
+#define TK_NK_SEMI       236
+#define TK_FILE          237
 
 #define TK_NK_SPACE   300
 #define TK_NK_COMMENT 301

include/libs/function/functionMgt.h

@@ -168,6 +168,9 @@ EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWin
 int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
 int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet);
 int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
+int32_t fmSetInvertFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
+int32_t fmSetNormalFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
+bool fmIsInvertible(int32_t funcId);
 
 #ifdef __cplusplus
 }

include/libs/function/tudf.h

@@ -44,7 +44,8 @@ enum {
   UDFC_CODE_PIPE_READ_ERR = -2,
   UDFC_CODE_CONNECT_PIPE_ERR = -3,
   UDFC_CODE_LOAD_UDF_FAILURE = -4,
-  UDFC_CODE_INVALID_STATE = -5
+  UDFC_CODE_INVALID_STATE = -5,
+  UDFC_CODE_NO_PIPE = -6,
 };
 
 typedef void *UdfcFuncHandle;
@@ -140,6 +141,44 @@ typedef int32_t (*TUdfDestroyFunc)();
 
 #define UDF_MEMORY_EXP_GROWTH 1.5
 
+#define udfColDataIsNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] == -1)
+#define udfColDataIsNull_f(pColumn, row) ((BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) & (1u << (7u - BitPos(row)))) == (1u << (7u - BitPos(row))))
+#define udfColDataSetNull_f(pColumn, row)                                                \
+  do {                                                                                   \
+    BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) |= (1u << (7u - BitPos(row))); \
+  } while (0)
+
+#define udfColDataSetNotNull_f(pColumn, r_)                                               \
+  do {                                                                                    \
+    BMCharPos(pColumn->colData.fixLenCol.nullBitmap, r_) &= ~(1u << (7u - BitPos(r_)));   \
+  } while (0)
+#define udfColDataSetNull_var(pColumn, row)  ((pColumn->colData.varLenCol.varOffsets)[row] = -1)
+
+
+static FORCE_INLINE char* udfColDataGetData(const SUdfColumn* pColumn, int32_t row) {
+  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
+    return pColumn->colData.varLenCol.payload + pColumn->colData.varLenCol.varOffsets[row];
+  } else {
+    return pColumn->colData.fixLenCol.data + pColumn->colMeta.bytes * row;
+  }
+}
+
+static FORCE_INLINE bool udfColDataIsNull(const SUdfColumn* pColumn, int32_t row) {
+  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
+    if (pColumn->colMeta.type == TSDB_DATA_TYPE_JSON) {
+      if (udfColDataIsNull_var(pColumn, row)) {
+        return true;
+      }
+      char* data = udfColDataGetData(pColumn, row);
+      return (*data == TSDB_DATA_TYPE_NULL);
+    } else {
+      return udfColDataIsNull_var(pColumn, row);
+    }
+  } else {
+    return udfColDataIsNull_f(pColumn, row);
+  }
+}
+
 static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t newCapacity) {
   SUdfColumnMeta *meta = &pColumn->colMeta;
   SUdfColumnData *data = &pColumn->colData;
@@ -186,17 +225,22 @@ static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t ne
   return TSDB_CODE_SUCCESS;
 }
 
-static FORCE_INLINE int32_t udfColSetRow(SUdfColumn* pColumn, uint32_t currentRow, const char* pData, bool isNull) {
+static FORCE_INLINE void udfColDataSetNull(SUdfColumn* pColumn, int32_t row) {
+  udfColEnsureCapacity(pColumn, row+1);
+  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
+    udfColDataSetNull_var(pColumn, row);
+  } else {
+    udfColDataSetNull_f(pColumn, row);
+  }
+}
+
+static FORCE_INLINE int32_t udfColDataSet(SUdfColumn* pColumn, uint32_t currentRow, const char* pData, bool isNull) {
   SUdfColumnMeta *meta = &pColumn->colMeta;
   SUdfColumnData *data = &pColumn->colData;
   udfColEnsureCapacity(pColumn, currentRow+1);
   bool isVarCol = IS_VAR_DATA_TYPE(meta->type);
   if (isNull) {
-    if (isVarCol) {
-      data->varLenCol.varOffsets[currentRow] = -1;
-    } else {
-      colDataSetNull_f(data->fixLenCol.nullBitmap, currentRow);
-    }
+      udfColDataSetNull(pColumn, currentRow);
   } else {
     if (!isVarCol) {
       colDataSetNotNull_f(data->fixLenCol.nullBitmap, currentRow);

include/libs/nodes/cmdnodes.h

@@ -310,6 +310,29 @@ typedef struct SCreateFunctionStmt {
   int32_t   bufSize;
 } SCreateFunctionStmt;
 
+typedef struct SDropFunctionStmt {
+  ENodeType type;
+  char      funcName[TSDB_FUNC_NAME_LEN];
+  bool      ignoreNotExists;
+} SDropFunctionStmt;
+
+#define PRIVILEGE_TYPE_MASK(n) (1 << n)
+
+#define PRIVILEGE_TYPE_ALL   PRIVILEGE_TYPE_MASK(0)
+#define PRIVILEGE_TYPE_READ  PRIVILEGE_TYPE_MASK(1)
+#define PRIVILEGE_TYPE_WRITE PRIVILEGE_TYPE_MASK(2)
+
+#define PRIVILEGE_TYPE_TEST_MASK(val, mask) (((val) & (mask)) != 0)
+
+typedef struct SGrantStmt {
+  ENodeType type;
+  char      userName[TSDB_USER_LEN];
+  char      dbName[TSDB_DB_NAME_LEN];
+  int64_t   privileges;
+} SGrantStmt;
+
+typedef SGrantStmt SRevokeStmt;
+
 #ifdef __cplusplus
 }
 #endif

include/libs/nodes/nodes.h

@@ -143,6 +143,8 @@ typedef enum ENodeType {
   QUERY_NODE_REDISTRIBUTE_VGROUP_STMT,
   QUERY_NODE_SPLIT_VGROUP_STMT,
   QUERY_NODE_SYNCDB_STMT,
+  QUERY_NODE_GRANT_STMT,
+  QUERY_NODE_REVOKE_STMT,
   QUERY_NODE_SHOW_DNODES_STMT,
   QUERY_NODE_SHOW_MNODES_STMT,
   QUERY_NODE_SHOW_MODULES_STMT,
@@ -174,8 +176,10 @@ typedef enum ENodeType {
   QUERY_NODE_SHOW_CREATE_DATABASE_STMT,
   QUERY_NODE_SHOW_CREATE_TABLE_STMT,
   QUERY_NODE_SHOW_CREATE_STABLE_STMT,
+  QUERY_NODE_SHOW_TRANSACTIONS_STMT,
   QUERY_NODE_KILL_CONNECTION_STMT,
   QUERY_NODE_KILL_QUERY_STMT,
+  QUERY_NODE_KILL_TRANSACTION_STMT,
 
   // logic plan node
   QUERY_NODE_LOGIC_PLAN_SCAN,

include/libs/planner/planner.h

@@ -25,20 +25,18 @@ extern "C" {
 
 typedef struct SPlanContext {
   uint64_t queryId;
-  int32_t acctId;
-  SEpSet mgmtEpSet;
-  SNode* pAstRoot;
-  bool topicQuery;
-  bool streamQuery;
-  bool rSmaQuery;
-  bool showRewrite;
-  int8_t triggerType;
-  int64_t watermark;
-  int32_t placeholderNum;
-  void* pTransporter;
-  struct SCatalog* pCatalog;
-  char* pMsg;
-  int32_t msgLen;
+  int32_t  acctId;
+  SEpSet   mgmtEpSet;
+  SNode*   pAstRoot;
+  bool     topicQuery;
+  bool     streamQuery;
+  bool     rSmaQuery;
+  bool     showRewrite;
+  int8_t   triggerType;
+  int64_t  watermark;
+  int32_t  placeholderNum;
+  char*    pMsg;
+  int32_t  msgLen;
 } SPlanContext;
 
 // Create the physical plan for the query, according to the AST.
@@ -47,7 +45,7 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
 // Set datasource of this subplan, multiple calls may be made to a subplan.
 // @pSubplan subplan to be schedule
 // @groupId id of a group of datasource subplans of this @pSubplan
-// @pSource one execution location of this group of datasource subplans 
+// @pSource one execution location of this group of datasource subplans
 int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
 
 int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId);
@@ -56,7 +54,7 @@ int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colI
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);
 int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan);
 
-char* qQueryPlanToString(const SQueryPlan* pPlan);
+char*       qQueryPlanToString(const SQueryPlan* pPlan);
 SQueryPlan* qStringToQueryPlan(const char* pStr);
 
 void qDestroyQueryPlan(SQueryPlan* pPlan);

include/libs/stream/tstream.h

@@ -69,20 +69,24 @@ typedef struct {
   SUseDbRsp dbInfo;
 } STaskDispatcherShuffle;
 
+typedef void FTbSink(void* vnode, int64_t ver, const SArray* data);
+
 typedef struct {
-  int8_t          reserved;
+  int64_t         stbUid;
   SSchemaWrapper* pSchemaWrapper;
   // not applicable to encoder and decoder
+  void*     vnode;
+  FTbSink*  tbSinkFunc;
   STSchema* pTSchema;
   SHashObj* pHash;  // groupId to tbuid
 } STaskSinkTb;
 
-typedef void FSmaHandle(void* vnode, int64_t smaId, const SArray* data);
+typedef void FSmaSink(void* vnode, int64_t smaId, const SArray* data);
 
 typedef struct {
   int64_t smaId;
   // following are not applicable to encoder and decoder
-  FSmaHandle* smaHandle;
+  FSmaSink* smaSink;
 } STaskSinkSma;
 
 typedef struct {
@@ -154,8 +158,8 @@ typedef struct {
 } SStreamTask;
 
 SStreamTask* tNewSStreamTask(int64_t streamId);
-int32_t      tEncodeSStreamTask(SCoder* pEncoder, const SStreamTask* pTask);
-int32_t      tDecodeSStreamTask(SCoder* pDecoder, SStreamTask* pTask);
+int32_t      tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask);
+int32_t      tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
 void         tFreeSStreamTask(SStreamTask* pTask);
 
 typedef struct {

include/libs/stream/tstreamUpdate.h

@@ -36,7 +36,7 @@ typedef struct SUpdateInfo {
 
 SUpdateInfo *updateInfoInitP(SInterval* pInterval, int64_t watermark);
 SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t watermark);
-bool isUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts);
+bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts);
 void updateInfoDestroy(SUpdateInfo *pInfo);
 
 #ifdef __cplusplus

include/util/taoserror.h

@@ -632,6 +632,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_FIRST_COLUMN      TAOS_DEF_ERROR_CODE(0, 0x2641)
 #define TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN    TAOS_DEF_ERROR_CODE(0, 0x2642)
 #define TSDB_CODE_PAR_INVALID_TAGS_NUM          TAOS_DEF_ERROR_CODE(0, 0x2643)
+#define TSDB_CODE_PAR_PERMISSION_DENIED         TAOS_DEF_ERROR_CODE(0, 0x2644)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

source/client/src/clientImpl.c

@@ -162,18 +162,17 @@ int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj*
 int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb) {
   STscObj* pTscObj = pRequest->pTscObj;
 
-  SParseContext cxt = {
-      .requestId = pRequest->requestId,
-      .acctId = pTscObj->acctId,
-      .db = pRequest->pDb,
-      .topicQuery = topicQuery,
-      .pSql = pRequest->sqlstr,
-      .sqlLen = pRequest->sqlLen,
-      .pMsg = pRequest->msgBuf,
-      .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
-      .pTransporter = pTscObj->pAppInfo->pTransporter,
-      .pStmtCb = pStmtCb,
-  };
+  SParseContext cxt = {.requestId = pRequest->requestId,
+                       .acctId = pTscObj->acctId,
+                       .db = pRequest->pDb,
+                       .topicQuery = topicQuery,
+                       .pSql = pRequest->sqlstr,
+                       .sqlLen = pRequest->sqlLen,
+                       .pMsg = pRequest->msgBuf,
+                       .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
+                       .pTransporter = pTscObj->pAppInfo->pTransporter,
+                       .pStmtCb = pStmtCb,
+                       .pUser = pTscObj->user};
 
   cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
   int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
@@ -232,11 +231,15 @@ int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArra
                       .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
                       .pAstRoot = pQuery->pRoot,
                       .showRewrite = pQuery->showRewrite,
-                      .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
                       .pMsg = pRequest->msgBuf,
                       .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                       .placeholderNum = pQuery->placeholderNum};
-  int32_t      code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
+  SEpSet       mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
+  SCatalog*    pCatalog = NULL;
+  int32_t      code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = catalogGetQnodeList(pCatalog, pRequest->pTscObj->pAppInfo->pTransporter, &mgmtEpSet, pNodeList);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
   }

source/client/src/tmq.c

@@ -547,21 +547,21 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
     req.offsets = (SMqOffset*)offsets->container.pData;
   }
 
-  SCoder encoder;
+  SEncoder encoder;
 
-  tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
+  tEncoderInit(&encoder, NULL, 0);
   tEncodeSMqCMCommitOffsetReq(&encoder, &req);
   int32_t tlen = encoder.pos;
   void*   buf = taosMemoryMalloc(tlen);
   if (buf == NULL) {
-    tCoderClear(&encoder);
+    tEncoderClear(&encoder);
     return -1;
   }
-  tCoderClear(&encoder);
+  tEncoderClear(&encoder);
 
-  tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, tlen, TD_ENCODER);
+  tEncoderInit(&encoder, buf, tlen);
   tEncodeSMqCMCommitOffsetReq(&encoder, &req);
-  tCoderClear(&encoder);
+  tEncoderClear(&encoder);
 
   pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_MQ_COMMIT_OFFSET);
   if (pRequest == NULL) {

source/common/src/tdatablock.c

@@ -363,9 +363,9 @@ int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc, SArray* pInd
 
   for (int32_t i = 0; i < pDest->info.numOfCols; ++i) {
     int32_t mapIndex = i;
-//    if (pIndexMap) {
-//      mapIndex = *(int32_t*)taosArrayGet(pIndexMap, i);
-//    }
+    //    if (pIndexMap) {
+    //      mapIndex = *(int32_t*)taosArrayGet(pIndexMap, i);
+    //    }
 
     SColumnInfoData* pCol2 = taosArrayGet(pDest->pDataBlock, i);
     SColumnInfoData* pCol1 = taosArrayGet(pSrc->pDataBlock, mapIndex);
@@ -1596,7 +1596,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
   return TSDB_CODE_SUCCESS;
 }
 
-SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema) {
+SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, bool createTb, int64_t suid) {
   SSubmitReq* ret = NULL;
 
   // cal size
@@ -1608,7 +1608,29 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema) {
     // TODO min
     int32_t rowSize = pDataBlock->info.rowSize;
     int32_t maxLen = TD_ROW_MAX_BYTES_FROM_SCHEMA(pTSchema);
-    cap += sizeof(SSubmitBlk) + rows * maxLen;
+    int32_t schemaLen = 0;
+
+    if (createTb) {
+      SVCreateTbReq createTbReq = {0};
+      createTbReq.name = "a";
+      createTbReq.flags = 0;
+      createTbReq.type = TSDB_CHILD_TABLE;
+      createTbReq.ctb.suid = htobe64(suid);
+
+      SKVRowBuilder kvRowBuilder = {0};
+      if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
+        ASSERT(0);
+      }
+      tdAddColToKVRow(&kvRowBuilder, 1, &pDataBlock->info.groupId, sizeof(uint64_t));
+      createTbReq.ctb.pTag = tdGetKVRowFromBuilder(&kvRowBuilder);
+      tdDestroyKVRowBuilder(&kvRowBuilder);
+
+      int32_t code;
+      tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
+      if (code < 0) return NULL;
+    }
+
+    cap += sizeof(SSubmitBlk) + schemaLen + rows * maxLen;
   }
 
   // assign data
@@ -1623,19 +1645,47 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema) {
 
     SSubmitBlk* blkHead = submitBlk;
     blkHead->numOfRows = htons(pDataBlock->info.rows);
-    blkHead->schemaLen = 0;
     blkHead->sversion = htonl(pTSchema->version);
     // TODO
-    blkHead->suid = 0;
-    blkHead->uid = htobe64(pDataBlock->info.uid);
+    blkHead->suid = htobe64(suid);
+    // uid is assigned by vnode
+    blkHead->uid = 0;
 
     int32_t rows = pDataBlock->info.rows;
     /*int32_t maxLen = TD_ROW_MAX_BYTES_FROM_SCHEMA(pTSchema);*/
     /*blkHead->dataLen = htonl(rows * maxLen);*/
     blkHead->dataLen = 0;
 
-    void*   blockData = POINTER_SHIFT(submitBlk, sizeof(SSubmitBlk));
-    STSRow* rowData = blockData;
+    void* blockData = POINTER_SHIFT(submitBlk, sizeof(SSubmitBlk));
+
+    int32_t schemaLen = 0;
+    if (createTb) {
+      SVCreateTbReq createTbReq = {0};
+      createTbReq.name = "a";
+      createTbReq.flags = 0;
+      createTbReq.type = TSDB_CHILD_TABLE;
+      createTbReq.ctb.suid = suid;
+
+      SKVRowBuilder kvRowBuilder = {0};
+      if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
+        ASSERT(0);
+      }
+      tdAddColToKVRow(&kvRowBuilder, 1, &pDataBlock->info.groupId, sizeof(uint64_t));
+      createTbReq.ctb.pTag = tdGetKVRowFromBuilder(&kvRowBuilder);
+      tdDestroyKVRowBuilder(&kvRowBuilder);
+
+      int32_t code;
+      tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
+      if (code < 0) return NULL;
+
+      SEncoder encoder = {0};
+      tEncoderInit(&encoder, blockData, schemaLen);
+      if (tEncodeSVCreateTbReq(&encoder, &createTbReq) < 0) return NULL;
+      tEncoderClear(&encoder);
+    }
+    blkHead->schemaLen = htonl(schemaLen);
+
+    STSRow* rowData = POINTER_SHIFT(blockData, schemaLen);
 
     for (int32_t j = 0; j < rows; j++) {
       SRowBuilder rb = {0};
@@ -1656,6 +1706,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema) {
     int32_t len = blkHead->dataLen;
     blkHead->dataLen = htonl(len);
     blkHead = POINTER_SHIFT(blkHead, len);
+    /*submitBlk = blkHead;*/
   }
 
   return ret;

source/dnode/mnode/impl/inc/mndDef.h

@@ -574,6 +574,7 @@ typedef struct {
   char           sourceDb[TSDB_DB_FNAME_LEN];
   char           targetDb[TSDB_DB_FNAME_LEN];
   char           targetSTbName[TSDB_TABLE_FNAME_LEN];
+  int64_t        targetStbUid;
   int64_t        createTime;
   int64_t        updateTime;
   int64_t        uid;
@@ -594,8 +595,8 @@ typedef struct {
   SSchemaWrapper outputSchema;
 } SStreamObj;
 
-int32_t tEncodeSStreamObj(SCoder* pEncoder, const SStreamObj* pObj);
-int32_t tDecodeSStreamObj(SCoder* pDecoder, SStreamObj* pObj);
+int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
+int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj);
 
 #ifdef __cplusplus
 }

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

@@ -411,11 +411,14 @@ void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) {
   return (void *)buf;
 }
 
-int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) {
+int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
   int32_t sz = 0;
   /*int32_t outputNameSz = 0;*/
   if (tEncodeCStr(pEncoder, pObj->name) < 0) return -1;
   if (tEncodeCStr(pEncoder, pObj->sourceDb) < 0) return -1;
+  if (tEncodeCStr(pEncoder, pObj->targetDb) < 0) return -1;
+  if (tEncodeCStr(pEncoder, pObj->targetSTbName) < 0) return -1;
+  if (tEncodeI64(pEncoder, pObj->targetStbUid) < 0) return -1;
   if (tEncodeI64(pEncoder, pObj->createTime) < 0) return -1;
   if (tEncodeI64(pEncoder, pObj->updateTime) < 0) return -1;
   if (tEncodeI64(pEncoder, pObj->uid) < 0) return -1;
@@ -462,9 +465,12 @@ int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) {
   return pEncoder->pos;
 }
 
-int32_t tDecodeSStreamObj(SCoder *pDecoder, SStreamObj *pObj) {
+int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj) {
   if (tDecodeCStrTo(pDecoder, pObj->name) < 0) return -1;
   if (tDecodeCStrTo(pDecoder, pObj->sourceDb) < 0) return -1;
+  if (tDecodeCStrTo(pDecoder, pObj->targetDb) < 0) return -1;
+  if (tDecodeCStrTo(pDecoder, pObj->targetSTbName) < 0) return -1;
+  if (tDecodeI64(pDecoder, &pObj->targetStbUid) < 0) return -1;
   if (tDecodeI64(pDecoder, &pObj->createTime) < 0) return -1;
   if (tDecodeI64(pDecoder, &pObj->updateTime) < 0) return -1;
   if (tDecodeI64(pDecoder, &pObj->uid) < 0) return -1;
@@ -529,4 +535,4 @@ void *tDecodeSMqOffsetObj(void *buf, SMqOffsetObj *pOffset) {
   buf = taosDecodeStringTo(buf, pOffset->key);
   buf = taosDecodeFixedI64(buf, &pOffset->offset);
   return buf;
-}
\ No newline at end of file
+}

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

@@ -157,8 +157,8 @@ static int32_t mndProcessCommitOffsetReq(SNodeMsg *pMsg) {
   SMnode              *pMnode = pMsg->pNode;
   char                *msgStr = pMsg->rpcMsg.pCont;
   SMqCMCommitOffsetReq commitOffsetReq;
-  SCoder               decoder;
-  tCoderInit(&decoder, TD_LITTLE_ENDIAN, msgStr, pMsg->rpcMsg.contLen, TD_DECODER);
+  SDecoder             decoder;
+  tDecoderInit(&decoder, msgStr, pMsg->rpcMsg.contLen);
 
   tDecodeSMqCMCommitOffsetReq(&decoder, &commitOffsetReq);
 

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

@@ -83,12 +83,12 @@ END:
 }
 
 int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet* pEpSet, tmsg_t type, int32_t nodeId) {
-  SCoder encoder;
-  tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
+  SEncoder encoder;
+  tEncoderInit(&encoder, NULL, 0);
   tEncodeSStreamTask(&encoder, pTask);
   int32_t size = encoder.pos;
   int32_t tlen = sizeof(SMsgHead) + size;
-  tCoderClear(&encoder);
+  tEncoderClear(&encoder);
   void* buf = taosMemoryMalloc(tlen);
   if (buf == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -96,9 +96,9 @@ int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet
   }
   ((SMsgHead*)buf)->vgId = htonl(nodeId);
   void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
-  tCoderInit(&encoder, TD_LITTLE_ENDIAN, abuf, size, TD_ENCODER);
+  tEncoderInit(&encoder, abuf, size);
   tEncodeSStreamTask(&encoder, pTask);
-  tCoderClear(&encoder);
+  tEncoderClear(&encoder);
 
   STransAction action = {0};
   memcpy(&action.epSet, pEpSet, sizeof(SEpSet));
@@ -204,6 +204,7 @@ int32_t mndAddShuffledSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* p
       pTask->smaSink.smaId = pStream->smaId;
     } else {
       pTask->sinkType = TASK_SINK__TABLE;
+      pTask->tbSink.stbUid = pStream->targetStbUid;
       pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
       ASSERT(pTask->tbSink.pSchemaWrapper);
     }
@@ -244,9 +245,10 @@ int32_t mndAddFixedSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStr
     pTask->smaSink.smaId = pStream->smaId;
   } else {
     pTask->sinkType = TASK_SINK__TABLE;
+    pTask->tbSink.stbUid = pStream->targetStbUid;
     pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
   }
-  //
+
   // dispatch
   pTask->dispatchType = TASK_DISPATCH__NONE;
 
@@ -319,6 +321,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
               pTask->smaSink.smaId = pStream->smaId;
             } else {
               pTask->sinkType = TASK_SINK__TABLE;
+              pTask->tbSink.stbUid = pStream->targetStbUid;
               pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
             }
 #endif

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

@@ -369,7 +369,7 @@ static FORCE_INLINE int schemaExColIdCompare(const void *colId, const void *pSch
 }
 
 static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
-  SCoder         coder = {0};
+  SEncoder       encoder = {0};
   int32_t        contLen;
   SName          name = {0};
   SVCreateStbReq req = {0};
@@ -422,11 +422,11 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
   pHead->vgId = htonl(pVgroup->vgId);
 
   void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pBuf, contLen - sizeof(SMsgHead), TD_ENCODER);
-  if (tEncodeSVCreateStbReq(&coder, &req) < 0) {
+  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
+  if (tEncodeSVCreateStbReq(&encoder, &req) < 0) {
     return NULL;
   }
-  tCoderClear(&coder);
+  tEncoderClear(&encoder);
 
   *pContLen = contLen;
   taosMemoryFreeClear(req.pRSmaParam.qmsg1);
@@ -440,7 +440,7 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb,
   int32_t      contLen = 0;
   int32_t      ret = 0;
   SMsgHead    *pHead = NULL;
-  SCoder       coder = {0};
+  SEncoder     encoder = {0};
 
   tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
 
@@ -462,9 +462,9 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb,
 
   void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
 
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pBuf, contLen - sizeof(SMsgHead), TD_ENCODER);
-  tEncodeSVDropStbReq(&coder, &req);
-  tCoderClear(&coder);
+  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
+  tEncodeSVDropStbReq(&encoder, &req);
+  tEncoderClear(&encoder);
 
   *pContLen = contLen;
   return pHead;

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

@@ -70,14 +70,14 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
   void *buf = NULL;
 
-  SCoder encoder;
-  tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
+  SEncoder encoder;
+  tEncoderInit(&encoder, NULL, 0);
   if (tEncodeSStreamObj(&encoder, pStream) < 0) {
-    tCoderClear(&encoder);
+    tEncoderClear(&encoder);
     goto STREAM_ENCODE_OVER;
   }
   int32_t tlen = encoder.pos;
-  tCoderClear(&encoder);
+  tEncoderClear(&encoder);
 
   int32_t  size = sizeof(int32_t) + tlen + MND_STREAM_RESERVE_SIZE;
   SSdbRaw *pRaw = sdbAllocRaw(SDB_STREAM, MND_STREAM_VER_NUMBER, size);
@@ -86,12 +86,12 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) {
   buf = taosMemoryMalloc(tlen);
   if (buf == NULL) goto STREAM_ENCODE_OVER;
 
-  tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, tlen, TD_ENCODER);
+  tEncoderInit(&encoder, buf, tlen);
   if (tEncodeSStreamObj(&encoder, pStream) < 0) {
-    tCoderClear(&encoder);
+    tEncoderClear(&encoder);
     goto STREAM_ENCODE_OVER;
   }
-  tCoderClear(&encoder);
+  tEncoderClear(&encoder);
 
   int32_t dataPos = 0;
   SDB_SET_INT32(pRaw, dataPos, tlen, STREAM_ENCODE_OVER);
@@ -138,8 +138,8 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
   if (buf == NULL) goto STREAM_DECODE_OVER;
   SDB_GET_BINARY(pRaw, dataPos, buf, tlen, STREAM_DECODE_OVER);
 
-  SCoder decoder;
-  tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, tlen + 1, TD_DECODER);
+  SDecoder decoder;
+  tDecoderInit(&decoder, buf, tlen + 1);
   if (tDecodeSStreamObj(&decoder, pStream) < 0) {
     goto STREAM_DECODE_OVER;
   }
@@ -360,6 +360,8 @@ static int32_t mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStre
     goto _OVER;
   }
 
+  stbObj.uid = pStream->targetStbUid;
+
   if (mndAddStbToTrans(pMnode, pTrans, pDb, &stbObj) < 0) goto _OVER;
 
   return 0;
@@ -379,6 +381,7 @@ static int32_t mndCreateStream(SMnode *pMnode, SNodeMsg *pReq, SCMCreateStreamRe
   streamObj.createTime = taosGetTimestampMs();
   streamObj.updateTime = streamObj.createTime;
   streamObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));
+  streamObj.targetStbUid = mndGenerateUid(pCreate->targetStbFullName, TSDB_TABLE_FNAME_LEN);
   streamObj.dbUid = pDb->uid;
   streamObj.version = 1;
   streamObj.sql = pCreate->sql;

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

@@ -412,7 +412,7 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
     goto _OVER;
   }
 
-  if (alterReq.pass[0] == 0) {
+  if (TSDB_ALTER_USER_PASSWD == alterReq.alterType && alterReq.pass[0] == 0) {
     terrno = TSDB_CODE_MND_INVALID_PASS_FORMAT;
     goto _OVER;
   }

source/dnode/snode/src/snode.c

@@ -99,10 +99,10 @@ void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
       ASSERT(0);
       return;
     }
-    SCoder decoder;
-    tCoderInit(&decoder, TD_LITTLE_ENDIAN, msg, pMsg->contLen - sizeof(SMsgHead), TD_DECODER);
+    SDecoder decoder;
+    tDecoderInit(&decoder, msg, pMsg->contLen - sizeof(SMsgHead));
     tDecodeSStreamTask(&decoder, pTask);
-    tCoderClear(&decoder);
+    tDecoderClear(&decoder);
 
     sndMetaDeployTask(pSnode->pMeta, pTask);
   } else if (pMsg->msgType == TDMT_SND_TASK_EXEC) {

source/dnode/vnode/inc/vnode.h

@@ -90,7 +90,7 @@ int         metaTbCursorNext(SMTbCursor *pTbCur);
 
 // tsdb
 // typedef struct STsdb STsdb;
-typedef void        *tsdbReaderT;
+typedef void *tsdbReaderT;
 
 #define BLOCK_LOAD_OFFSET_SEQ_ORDER 1
 #define BLOCK_LOAD_TABLE_SEQ_ORDER  2
@@ -108,12 +108,12 @@ int32_t      tsdbQuerySTableByTagCond(void *pMeta, uint64_t uid, TSKEY skey, con
 int64_t      tsdbGetNumOfRowsInMemTable(tsdbReaderT *pHandle);
 bool         tsdbNextDataBlock(tsdbReaderT pTsdbReadHandle);
 void         tsdbRetrieveDataBlockInfo(tsdbReaderT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
-int32_t      tsdbRetrieveDataBlockStatisInfo(tsdbReaderT *pTsdbReadHandle, SColumnDataAgg ***pBlockStatis, bool* allHave);
-SArray      *tsdbRetrieveDataBlock(tsdbReaderT *pTsdbReadHandle, SArray *pColumnIdList);
-void         tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond);
-void         tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
-int32_t      tsdbGetOneTableGroup(void *pMeta, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo);
-int32_t      tsdbGetTableGroupFromIdList(SVnode *pVnode, SArray *pTableIdList, STableGroupInfo *pGroupInfo);
+int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT *pTsdbReadHandle, SColumnDataAgg ***pBlockStatis, bool *allHave);
+SArray *tsdbRetrieveDataBlock(tsdbReaderT *pTsdbReadHandle, SArray *pColumnIdList);
+void    tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond);
+void    tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
+int32_t tsdbGetOneTableGroup(void *pMeta, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo);
+int32_t tsdbGetTableGroupFromIdList(SVnode *pVnode, SArray *pTableIdList, STableGroupInfo *pGroupInfo);
 
 // tq
 
@@ -126,8 +126,8 @@ int     tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList
 int     tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
 int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
 bool    tqNextDataBlock(STqReadHandle *pHandle);
-int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t* pUid, int32_t *pNumOfRows,
-                            int16_t *pNumOfCols);
+int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t *pUid,
+                            int32_t *pNumOfRows, int16_t *pNumOfCols);
 
 // need to reposition
 
@@ -189,10 +189,10 @@ struct SMetaEntry {
       SSchemaWrapper schemaTag;
     } stbEntry;
     struct {
-      int64_t     ctime;
-      int32_t     ttlDays;
-      tb_uid_t    suid;
-      const void *pTags;
+      int64_t        ctime;
+      int32_t        ttlDays;
+      tb_uid_t       suid;
+      const uint8_t *pTags;
     } ctbEntry;
     struct {
       int64_t        ctime;
@@ -205,7 +205,7 @@ struct SMetaEntry {
 struct SMetaReader {
   int32_t    flags;
   SMeta     *pMeta;
-  SCoder     coder;
+  SDecoder   coder;
   SMetaEntry me;
   void      *pBuf;
   int        szBuf;

source/dnode/vnode/src/inc/meta.h

@@ -39,8 +39,8 @@ typedef struct SMSmaCursor SMSmaCursor;
 // metaOpen ==================
 
 // metaEntry ==================
-int metaEncodeEntry(SCoder* pCoder, const SMetaEntry* pME);
-int metaDecodeEntry(SCoder* pCoder, SMetaEntry* pME);
+int metaEncodeEntry(SEncoder* pCoder, const SMetaEntry* pME);
+int metaDecodeEntry(SDecoder* pCoder, SMetaEntry* pME);
 
 // metaTable ==================
 

source/dnode/vnode/src/meta/metaEntry.c

@@ -15,7 +15,7 @@
 
 #include "meta.h"
 
-int metaEncodeEntry(SCoder *pCoder, const SMetaEntry *pME) {
+int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) {
   if (tStartEncode(pCoder) < 0) return -1;
 
   if (tEncodeI64(pCoder, pME->version) < 0) return -1;
@@ -43,8 +43,8 @@ int metaEncodeEntry(SCoder *pCoder, const SMetaEntry *pME) {
   return 0;
 }
 
-int metaDecodeEntry(SCoder *pCoder, SMetaEntry *pME) {
-  uint64_t len;
+int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
+  uint32_t len;
   if (tStartDecode(pCoder) < 0) return -1;
 
   if (tDecodeI64(pCoder, &pME->version) < 0) return -1;

source/dnode/vnode/src/meta/metaQuery.c

@@ -22,7 +22,7 @@ void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) {
 }
 
 void metaReaderClear(SMetaReader *pReader) {
-  tCoderClear(&pReader->coder);
+  tDecoderClear(&pReader->coder);
   tdbFree(pReader->pBuf);
 }
 
@@ -37,7 +37,7 @@ int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t u
   }
 
   // decode the entry
-  tCoderInit(&pReader->coder, TD_LITTLE_ENDIAN, pReader->pBuf, pReader->szBuf, TD_DECODER);
+  tDecoderInit(&pReader->coder, pReader->pBuf, pReader->szBuf);
 
   if (metaDecodeEntry(&pReader->coder, &pReader->me) < 0) {
     goto _err;
@@ -147,7 +147,7 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
   SSchemaWrapper *pSW = NULL;
   SSchema        *pSchema = NULL;
   void           *pBuf;
-  SCoder          coder = {0};
+  SDecoder        coder = {0};
 
   // fetch
   skmDbKey.uid = uid;
@@ -163,11 +163,11 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
   pBuf = pVal;
   pSW = taosMemoryMalloc(sizeof(SSchemaWrapper));
 
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_DECODER);
+  tDecoderInit(&coder, pVal, vLen);
   tDecodeSSchemaWrapper(&coder, pSW);
   pSchema = taosMemoryMalloc(sizeof(SSchema) * pSW->nCols);
   memcpy(pSchema, pSW->pSchema, sizeof(SSchema) * pSW->nCols);
-  tCoderClear(&coder);
+  tDecoderClear(&coder);
 
   pSW->pSchema = pSchema;
 

source/dnode/vnode/src/meta/metaTable.c

@@ -33,7 +33,6 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   void       *pBuf = NULL;
   int32_t     szBuf = 0;
   void       *p = NULL;
-  SCoder      coder = {0};
   SMetaReader mr = {0};
 
   // validate req
@@ -192,7 +191,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
   tb_uid_t    uid;
   int64_t     tver;
   SMetaEntry  me = {0};
-  SCoder      coder = {0};
+  SDecoder    coder = {0};
   int8_t      type;
   int64_t     ctime;
   tb_uid_t    suid;
@@ -253,7 +252,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
   // decode entry
   void *pDataCopy = taosMemoryMalloc(nData);  // remove the copy (todo)
   memcpy(pDataCopy, pData, nData);
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pDataCopy, nData, TD_DECODER);
+  tDecoderInit(&coder, pDataCopy, nData);
   ret = metaDecodeEntry(&coder, &me);
   if (ret < 0) {
     ASSERT(0);
@@ -272,7 +271,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
   }
 
   taosMemoryFree(pDataCopy);
-  tCoderClear(&coder);
+  tDecoderClear(&coder);
   tdbDbcClose(pTbDbc);
 
   if (type == TSDB_CHILD_TABLE) {
@@ -309,7 +308,7 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   void    *pVal = NULL;
   int      kLen = 0;
   int      vLen = 0;
-  SCoder   coder = {0};
+  SEncoder coder = {0};
 
   // set key and value
   tbDbKey.version = pME->version;
@@ -330,13 +329,13 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
     goto _err;
   }
 
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_ENCODER);
+  tEncoderInit(&coder, pVal, vLen);
 
   if (metaEncodeEntry(&coder, pME) < 0) {
     goto _err;
   }
 
-  tCoderClear(&coder);
+  tEncoderClear(&coder);
 
   // write to table.db
   if (tdbDbInsert(pMeta->pTbDb, pKey, kLen, pVal, vLen, &pMeta->txn) < 0) {
@@ -393,7 +392,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pME) {
 }
 
 static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
-  SCoder                coder = {0};
+  SEncoder              coder = {0};
   void                 *pVal = NULL;
   int                   vLen = 0;
   int                   rcode = 0;
@@ -422,7 +421,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
     goto _exit;
   }
 
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_ENCODER);
+  tEncoderInit(&coder, pVal, vLen);
   tEncodeSSchemaWrapper(&coder, pSW);
 
   if (tdbDbInsert(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), pVal, vLen, &pMeta->txn) < 0) {
@@ -432,7 +431,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
 
 _exit:
   taosMemoryFree(pVal);
-  tCoderClear(&coder);
+  tEncoderClear(&coder);
   return rcode;
 }
 

source/dnode/vnode/src/tq/tq.c

@@ -884,24 +884,38 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
   }
 }
 
-int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int32_t parallel) {
-  if (pTask->execType == TASK_EXEC__NONE) return 0;
+void tqTableSink(void* vnode, int64_t ver, const SArray* data) {
+  //
+  SVnode* pVnode = (SVnode*)vnode;
+  // build write msg
+  //
+}
 
-  pTask->exec.numOfRunners = parallel;
-  pTask->exec.runners = taosMemoryCalloc(parallel, sizeof(SStreamRunner));
-  if (pTask->exec.runners == NULL) {
-    return -1;
+int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int32_t parallel) {
+  if (pTask->execType != TASK_EXEC__NONE) {
+    // expand runners
+    pTask->exec.numOfRunners = parallel;
+    pTask->exec.runners = taosMemoryCalloc(parallel, sizeof(SStreamRunner));
+    if (pTask->exec.runners == NULL) {
+      return -1;
+    }
+    for (int32_t i = 0; i < parallel; i++) {
+      STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
+      SReadHandle    handle = {
+             .reader = pStreamReader,
+             .meta = pTq->pVnode->pMeta,
+      };
+      pTask->exec.runners[i].inputHandle = pStreamReader;
+      pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
+      ASSERT(pTask->exec.runners[i].executor);
+    }
   }
-  for (int32_t i = 0; i < parallel; i++) {
-    STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
-    SReadHandle    handle = {
-           .reader = pStreamReader,
-           .meta = pTq->pVnode->pMeta,
-    };
-    pTask->exec.runners[i].inputHandle = pStreamReader;
-    pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
-    ASSERT(pTask->exec.runners[i].executor);
+
+  if (pTask->sinkType == TASK_SINK__TABLE) {
+    pTask->tbSink.vnode = pTq->pVnode;
+    pTask->tbSink.tbSinkFunc = tqTableSink;
   }
+
   return 0;
 }
 
@@ -910,12 +924,12 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
   if (pTask == NULL) {
     return -1;
   }
-  SCoder decoder;
-  tCoderInit(&decoder, TD_LITTLE_ENDIAN, (uint8_t*)msg, msgLen, TD_DECODER);
+  SDecoder decoder;
+  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
   if (tDecodeSStreamTask(&decoder, pTask) < 0) {
     ASSERT(0);
   }
-  tCoderClear(&decoder);
+  tDecoderClear(&decoder);
 
   // exec
   if (tqExpandTask(pTq, pTask, 4) < 0) {
@@ -925,7 +939,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
   // sink
   pTask->ahandle = pTq->pVnode;
   if (pTask->sinkType == TASK_SINK__SMA) {
-    pTask->smaSink.smaHandle = smaHandleRes;
+    pTask->smaSink.smaSink = smaHandleRes;
   } else if (pTask->sinkType == TASK_SINK__TABLE) {
     ASSERT(pTask->tbSink.pSchemaWrapper);
     ASSERT(pTask->tbSink.pSchemaWrapper->pSchema);

source/dnode/vnode/src/tsdb/tsdbMemTable.c

@@ -249,8 +249,10 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
             pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, rowKey);
             pMergeInfo->keyLast = TMAX(pMergeInfo->keyLast, rowKey);
             lastKey = rowKey;
-            ++pCols->numOfRows;
-            tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false);
+            if (pCols) {
+              ++pCols->numOfRows;
+              tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false);
+            }
           } else {
             tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, true);
           }
@@ -279,7 +281,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
     }
 #endif
   }
-  if (lastKey != TSKEY_INITIAL_VAL) {
+  if (pCols && (lastKey != TSKEY_INITIAL_VAL)) {
     ++pCols->numOfRows;
   }
 

source/dnode/vnode/src/tsdb/tsdbSma.c

@@ -1638,7 +1638,7 @@ int32_t tsdbCreateTSma(STsdb *pTsdb, char *pMsg) {
     tsdbWarn("vgId:%d tsma create msg received but deserialize failed since %s", REPO_ID(pTsdb), terrstr(terrno));
     return -1;
   }
-  
+
   tsdbDebug("vgId:%d tsma create msg %s:%" PRIi64 " for table %" PRIi64 " received", REPO_ID(pTsdb),
             vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid);
 
@@ -2006,6 +2006,12 @@ static FORCE_INLINE int32_t tsdbExecuteRSmaImpl(STsdb *pTsdb, const void *pMsg,
                                                 qTaskInfo_t *taskInfo, STSchema *pTSchema, tb_uid_t suid, tb_uid_t uid,
                                                 int8_t level) {
   SArray *pResult = NULL;
+
+  if (!taskInfo) {
+    tsdbDebug("vgId:%d no qTaskInfo to execute rsma %" PRIi8 " task for suid:%" PRIu64, REPO_ID(pTsdb), level, suid);
+    return TSDB_CODE_SUCCESS;
+  }
+
   tsdbDebug("vgId:%d execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, REPO_ID(pTsdb), level, taskInfo,
             suid);
 
@@ -2071,10 +2077,18 @@ static int32_t tsdbExecuteRSma(STsdb *pTsdb, const void *pMsg, int32_t inputType
     tsdbDebug("vgId:%d no rsma info for suid:%" PRIu64, REPO_ID(pTsdb), suid);
     return TSDB_CODE_SUCCESS;
   }
+  if (!pRSmaInfo->taskInfo[0]) {
+    tsdbDebug("vgId:%d no rsma qTaskInfo for suid:%" PRIu64, REPO_ID(pTsdb), suid);
+    return TSDB_CODE_SUCCESS;
+  }
 
   if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
     // TODO: use the proper schema instead of 0, and cache STSchema in cache
     STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, suid, 0);
+    if (!pTSchema) {
+      terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
+      return TSDB_CODE_FAILED;
+    }
     tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[0], pTSchema, suid, uid, TSDB_RETENTION_L1);
     tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[1], pTSchema, suid, uid, TSDB_RETENTION_L2);
     taosMemoryFree(pTSchema);

source/dnode/vnode/src/vnd/vnodeCfg.c

@@ -25,7 +25,7 @@ const SVnodeCfg vnodeCfgDefault = {
     .isHeap = false,
     .isWeak = 0,
     .tsdbCfg = {.precision = TSDB_TIME_PRECISION_MILLI,
-                .update = 0,
+                .update = 1,
                 .compression = 2,
                 .slLevel = 5,
                 .days = 10,

source/libs/executor/CMakeLists.txt

@@ -17,6 +17,6 @@ target_include_directories(
         PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 
-#if(${BUILD_TEST})
+if(${BUILD_TEST})
     ADD_SUBDIRECTORY(test)
-#endif(${BUILD_TEST})
+endif(${BUILD_TEST})

source/libs/executor/src/executorimpl.c

@@ -198,6 +198,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
 
   pBlock->info.blockId = pNode->dataBlockId;
   pBlock->info.rowSize = pNode->totalRowSize;  // todo ??
+  pBlock->info.type = STREAM_INVALID;
 
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData idata = {{0}};

source/libs/function/inc/builtins.h

@@ -36,6 +36,7 @@ typedef struct SBuiltinFuncDefinition {
   FExecProcess processFunc;
   FScalarExecProcess sprocessFunc;
   FExecFinalize finalizeFunc;
+  FExecProcess invertFunc;
 } SBuiltinFuncDefinition;
 
 extern const SBuiltinFuncDefinition funcMgtBuiltins[];

source/libs/parser/inc/parInt.h

@@ -26,6 +26,7 @@ extern "C" {
 
 int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery);
 int32_t parse(SParseContext* pParseCxt, SQuery** pQuery);
+int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery);
 int32_t translate(SParseContext* pParseCxt, SQuery* pQuery);
 int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
 int32_t calculateConstant(SParseContext* pParseCxt, SQuery* pQuery);