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

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/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/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

@@ -154,8 +154,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/dnode/mnode/impl/inc/mndDef.h

@@ -594,8 +594,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,7 +411,7 @@ 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;
@@ -462,7 +462,7 @@ 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 (tDecodeI64(pDecoder, &pObj->createTime) < 0) return -1;

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));

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;
   }

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

@@ -913,12 +913,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) {

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

@@ -63,14 +63,18 @@ struct SMemSkipListCurosr {
   SMemSkipListNode *pNodeC;
 };
 
+#define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET))
+
 #define SL_NODE_SIZE(l)        (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
 #define SL_NODE_HALF_SIZE(l)   (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l))
 #define SL_NODE_FORWARD(n, l)  ((n)->forwards[l])
 #define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
 #define SL_NODE_DATA(n)        (&SL_NODE_BACKWARD(n, (n)->level))
 
-#define SL_HEAD_NODE(sl) ((sl)->pHead)
-#define SL_TAIL_NODE(sl) ((SMemSkipListNode *)&SL_NODE_FORWARD(SL_HEAD_NODE(sl), (sl)->maxLevel))
+#define SL_HEAD_NODE(sl)            ((sl)->pHead)
+#define SL_TAIL_NODE(sl)            ((SMemSkipListNode *)&SL_NODE_FORWARD(SL_HEAD_NODE(sl), (sl)->maxLevel))
+#define SL_HEAD_NODE_FORWARD(n, l)  SL_NODE_FORWARD(n, l)
+#define SL_TAIL_NODE_BACKWARD(n, l) SL_NODE_FORWARD(n, l)
 
 // SMemTable
 int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
@@ -111,23 +115,18 @@ int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMemTb) {
 }
 
 int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *pSubmitBlk) {
-  SMemData          *pMemData;
-  STsdb             *pTsdb = pMemTb->pTsdb;
-  SVnode            *pVnode = pTsdb->pVnode;
-  SVBufPool         *pPool = pVnode->inUse;
-  int32_t            hash;
-  int32_t            tlen;
-  uint8_t            buf[16];
-  int32_t            rlen;
-  const uint8_t     *p;
-  SMemSkipListNode  *pSlNode;
-  const STSRow      *pTSRow;
-  SMemSkipListCurosr slc = {0};
-
-  // search hash
-  hash = (pSubmitBlk->suid + pSubmitBlk->uid) % pMemTb->nBucket;
-  for (pMemData = pMemTb->pBuckets[hash]; pMemData; pMemData = pMemData->pHashNext) {
-    if (pMemData->suid == pSubmitBlk->suid && pMemData->uid == pSubmitBlk->uid) break;
+  SMemData  *pMemData;
+  STsdb     *pTsdb = pMemTb->pTsdb;
+  SVnode    *pVnode = pTsdb->pVnode;
+  SVBufPool *pPool = pVnode->inUse;
+  tb_uid_t   suid = pSubmitBlk->suid;
+  tb_uid_t   uid = pSubmitBlk->uid;
+  int32_t    iBucket;
+
+  // search SMemData by hash
+  iBucket = HASH_BUCKET(suid, uid, pMemTb->nBucket);
+  for (pMemData = pMemTb->pBuckets[iBucket]; pMemData; pMemData = pMemData->pHashNext) {
+    if (pMemData->suid == suid && pMemData->uid == uid) break;
   }
 
   // create pMemData if need
@@ -143,8 +142,8 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
     }
 
     pMemData->pHashNext = NULL;
-    pMemData->suid = pSubmitBlk->suid;
-    pMemData->uid = pSubmitBlk->uid;
+    pMemData->suid = suid;
+    pMemData->uid = uid;
     pMemData->minKey = TSKEY_MAX;
     pMemData->maxKey = TSKEY_MIN;
     pMemData->minVer = -1;
@@ -159,55 +158,67 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
     pHead->level = maxLevel;
     pTail->level = maxLevel;
     for (int iLevel = 0; iLevel < maxLevel; iLevel++) {
-      SL_NODE_FORWARD(pHead, iLevel) = pTail;
-      SL_NODE_FORWARD(pTail, iLevel) = pHead;
+      SL_HEAD_NODE_FORWARD(pHead, iLevel) = pTail;
+      SL_TAIL_NODE_BACKWARD(pTail, iLevel) = pHead;
     }
 
-    // add to MemTable
-    hash = (pMemData->suid + pMemData->uid) % pMemTb->nBucket;
-    pMemData->pHashNext = pMemTb->pBuckets[hash];
-    pMemTb->pBuckets[hash] = pMemData;
+    // add to hash
+    if (pMemTb->nHash >= pMemTb->nBucket) {
+      // rehash (todo)
+    }
+    iBucket = HASH_BUCKET(suid, uid, pMemTb->nBucket);
+    pMemData->pHashNext = pMemTb->pBuckets[iBucket];
+    pMemTb->pBuckets[iBucket] = pMemData;
     pMemTb->nHash++;
+
+    // sort organize (todo)
   }
 
-// loop to insert data to skiplist
-#if 0
-  tsdbMemSkipListCursorOpen(&slc, &pMemData->sl);
-  p = pSubmitBlk->pData;
-  for (;;) {
-    if (p - (uint8_t *)pSubmitBlk->pData >= pSubmitBlk->nData) break;
+  // do insert data to SMemData
+  SMemSkipListCurosr slc = {0};
+  const uint8_t     *p = pSubmitBlk->pData;
+  const uint8_t     *pt;
+  const STSRow      *pRow;
+  uint64_t           szRow;
+  SDecoder           decoder = {0};
 
-    const uint8_t *pt = p;
-    p = tGetBinary(p, &pTSRow, &rlen);
+  // tCoderInit(&coder, TD_LITTLE_ENDIAN, pSubmitBlk->pData, pSubmitBlk->nData, TD_DECODER);
+  for (;;) {
+    // if (tDecodeIsEnd(&coder)) break;
 
+    // if (tDecodeBinary(&coder, (const uint8_t **)&pRow, &szRow) < 0) {
+    //   terrno = TSDB_CODE_INVALID_MSG;
+    //   return -1;
+    // }
     // check the row (todo)
 
-    // move the cursor to position to write (todo)
-    int32_t c;
-    tsdbMemSkipListCursorMoveTo(&slc, pTSRow, version, &c);
-    ASSERT(c);
+    //   // move the cursor to position to write (todo)
+    //   int32_t c;
+    //   tsdbMemSkipListCursorMoveTo(&slc, pTSRow, version, &c);
+    //   ASSERT(c);
 
-    // encode row
-    int8_t  level = tsdbMemSkipListRandLevel(&pMemData->sl);
-    int32_t tsize = SL_NODE_SIZE(level) + sizeof(version) + (p - pt);
-    pSlNode = vnodeBufPoolMalloc(pPool, tsize);
-    pSlNode->level = level;
+    //   // encode row
+    //   int8_t  level = tsdbMemSkipListRandLevel(&pMemData->sl);
+    //   int32_t tsize = SL_NODE_SIZE(level) + sizeof(version) + (p - pt);
+    //   pSlNode = vnodeBufPoolMalloc(pPool, tsize);
+    //   pSlNode->level = level;
 
-    uint8_t *pData = SL_NODE_DATA(pSlNode);
-    *(int64_t *)pData = version;
-    pData += sizeof(version);
-    memcpy(pData, pt, p - pt);
+    //   uint8_t *pData = SL_NODE_DATA(pSlNode);
+    //   *(int64_t *)pData = version;
+    //   pData += sizeof(version);
+    //   memcpy(pData, pt, p - pt);
 
-    // insert row
-    tsdbMemSkipListCursorPut(&slc, pSlNode);
+    //   // insert row
+    //   tsdbMemSkipListCursorPut(&slc, pSlNode);
 
     // update status
-    if (pTSRow->ts < pMemData->minKey) pMemData->minKey = pTSRow->ts;
-    if (pTSRow->ts > pMemData->maxKey) pMemData->maxKey = pTSRow->ts;
+    if (pRow->ts < pMemData->minKey) pMemData->minKey = pRow->ts;
+    if (pRow->ts > pMemData->maxKey) pMemData->maxKey = pRow->ts;
   }
-  tsdbMemSkipListCursorClose(&slc);
-#endif
+  // tCoderClear(&coder);
+  // tsdbMemSkipListCursorClose(&slc);
 
+  // update status
   if (pMemData->minVer == -1) pMemData->minVer = version;
   if (pMemData->maxVer == -1 || pMemData->maxVer < version) pMemData->maxVer = version;
 
@@ -217,8 +228,4 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
   if (pMemTb->maxVer == -1 || pMemTb->maxVer < version) pMemTb->maxVer = version;
 
   return 0;
-}
-
+}
\ No newline at end of file
-// SMemData
-
-// SMemSkipList
\ No newline at end of file

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

@@ -286,7 +286,7 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
 
 static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) {
   SVCreateStbReq req = {0};
-  SCoder         coder;
+  SDecoder       coder;
 
   pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
   pRsp->code = TSDB_CODE_SUCCESS;
@@ -294,7 +294,7 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq,
   pRsp->contLen = 0;
 
   // decode and process req
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER);
+  tDecoderInit(&coder, pReq, len);
 
   if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
     pRsp->code = terrno;
@@ -308,16 +308,16 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq,
 
   tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &req, &pVnode->msgCb);
 
-  tCoderClear(&coder);
+  tDecoderClear(&coder);
   return 0;
 
 _err:
-  tCoderClear(&coder);
+  tDecoderClear(&coder);
   return -1;
 }
 
 static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) {
-  SCoder             coder = {0};
+  SDecoder           decoder = {0};
   int                rcode = 0;
   SVCreateTbBatchReq req = {0};
   SVCreateTbReq     *pCreateReq;
@@ -332,8 +332,8 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
   pRsp->contLen = 0;
 
   // decode
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER);
-  if (tDecodeSVCreateTbBatchReq(&coder, &req) < 0) {
+  tDecoderInit(&decoder, pReq, len);
+  if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {
     rcode = -1;
     terrno = TSDB_CODE_INVALID_MSG;
     goto _exit;
@@ -373,13 +373,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
     taosArrayPush(rsp.pArray, &cRsp);
   }
 
-  tCoderClear(&coder);
+  tDecoderClear(&decoder);
 
   tsdbUpdateTbUidList(pVnode->pTsdb, pStore);
   tsdbUidStoreFree(pStore);
 
   // prepare rsp
-  int32_t ret = 0;
+  SEncoder encoder = {0};
+  int32_t  ret = 0;
   tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);
   pRsp->pCont = rpcMallocCont(pRsp->contLen);
   if (pRsp->pCont == NULL) {
@@ -387,12 +388,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
     rcode = -1;
     goto _exit;
   }
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pRsp->pCont, pRsp->contLen, TD_ENCODER);
-  tEncodeSVCreateTbBatchRsp(&coder, &rsp);
+  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
+  tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
+  tEncoderClear(&encoder);
 
 _exit:
   taosArrayClear(rsp.pArray);
-  tCoderClear(&coder);
+  tDecoderClear(&decoder);
+  tEncoderClear(&encoder);
   return rcode;
 }
 
@@ -416,15 +419,15 @@ static int vnodeProcessAlterStbReq(SVnode *pVnode, void *pReq, int32_t len, SRpc
 static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
   SVDropStbReq req = {0};
   int          rcode = TSDB_CODE_SUCCESS;
-  SCoder       coder = {0};
+  SDecoder     decoder = {0};
 
   pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
   pRsp->pCont = NULL;
   pRsp->contLen = 0;
 
   // decode request
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER);
-  if (tDecodeSVDropStbReq(&coder, &req) < 0) {
+  tDecoderInit(&decoder, pReq, len);
+  if (tDecodeSVDropStbReq(&decoder, &req) < 0) {
     rcode = TSDB_CODE_INVALID_MSG;
     goto _exit;
   }
@@ -438,7 +441,7 @@ static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, i
   // return rsp
 _exit:
   pRsp->code = rcode;
-  tCoderClear(&coder);
+  tDecoderClear(&decoder);
   return 0;
 }
 
@@ -451,7 +454,7 @@ static int vnodeProcessAlterTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcM
 static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
   SVDropTbBatchReq req = {0};
   SVDropTbBatchRsp rsp = {0};
-  SCoder           coder = {0};
+  SDecoder         decoder = {0};
   int              ret;
 
   pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
@@ -460,8 +463,8 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
   pRsp->code = TSDB_CODE_SUCCESS;
 
   // decode req
-  tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER);
-  ret = tDecodeSVDropTbBatchReq(&coder, &req);
+  tDecoderInit(&decoder, pReq, len);
+  ret = tDecodeSVDropTbBatchReq(&decoder, &req);
   if (ret < 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     pRsp->code = terrno;
@@ -490,7 +493,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
   }
 
 _exit:
-  tCoderClear(&coder);
+  tDecoderClear(&decoder);
   // encode rsp (TODO)
   return 0;
 }
@@ -501,7 +504,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
   SSubmitBlk    *pBlock;
   SSubmitRsp     rsp = {0};
   SVCreateTbReq  createTbReq = {0};
-  SCoder         coder = {0};
+  SDecoder       decoder = {0};
   int32_t        nRows;
 
   pRsp->code = 0;
@@ -518,17 +521,17 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
 
     // create table for auto create table mode
     if (msgIter.schemaLen > 0) {
-      tCoderInit(&coder, TD_LITTLE_ENDIAN, pBlock->data, msgIter.schemaLen, TD_DECODER);
-      if (tDecodeSVCreateTbReq(&coder, &createTbReq) < 0) {
+      tDecoderInit(&decoder, pBlock->data, msgIter.schemaLen);
+      if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) {
         pRsp->code = TSDB_CODE_INVALID_MSG;
-        tCoderClear(&coder);
+        tDecoderClear(&decoder);
         goto _exit;
       }
 
       if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) {
         if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
           pRsp->code = terrno;
-          tCoderClear(&coder);
+          tDecoderClear(&decoder);
           goto _exit;
         }
       }
@@ -540,7 +543,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
         msgIter.suid = 0;
       }
 
-      tCoderClear(&coder);
+      tDecoderClear(&decoder);
     }
 
     if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &nRows) < 0) {

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/inc/executorimpl.h

@@ -38,6 +38,7 @@ extern "C" {
 #include "tlockfree.h"
 #include "tmsg.h"
 #include "tpagedbuf.h"
+#include "tstreamUpdate.h"
 
 #include "vnode.h"
 #include "executorInt.h"
@@ -225,7 +226,6 @@ typedef struct SExecTaskInfo {
   char*            sql;                  // query sql string
   jmp_buf          env;                  // jump to this position when error happens.
   EOPTR_EXEC_MODEL execModel;            // operator execution model [batch model|stream model]
-  struct SSubplan  *plan;
   struct SOperatorInfo* pRoot;
 } SExecTaskInfo;
 
@@ -387,6 +387,9 @@ typedef struct SStreamBlockScanInfo {
   void*        readerHandle;     // stream block reader handle
   SArray*      pColMatchInfo;    //
   SNode*       pCondition;
+  SArray*      tsArray;
+  SUpdateInfo*  pUpdateInfo;
+  int32_t      primaryTsIndex;    // primary time stamp slot id
 } SStreamBlockScanInfo;
 
 typedef struct SSysTableScanInfo {
@@ -444,6 +447,7 @@ typedef struct SIntervalAggOperatorInfo {
   SArray*            pUpdatedWindow;     // updated time window due to the input data block from the downstream operator.
   STimeWindowAggSupp twAggSup;
   struct SFillInfo*  pFillInfo;          // fill info
+  bool               invertible;
 } SIntervalAggOperatorInfo;
 
 typedef struct SAggOperatorInfo {

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}};
@@ -5189,7 +5190,6 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead
     goto _complete;
   }
 
-  (*pTaskInfo)->plan = pPlan;
   return code;
 
 _complete:

source/libs/executor/src/scanoperator.c

@@ -512,6 +512,25 @@ static void doClearBufferedBlocks(SStreamBlockScanInfo* pInfo) {
   taosArrayClear(pInfo->pBlockLists);
 }
 
+static SSDataBlock* getUpdateDataBlock(SStreamBlockScanInfo* pInfo) {
+  SColumnInfoData* pColDataInfo = taosArrayGet(pInfo->pRes->pDataBlock, pInfo->primaryTsIndex);
+  TSKEY* ts = (TSKEY*)pColDataInfo->pData;
+  for (int32_t i = 0; i < pInfo->pRes->info.rows; i++) {
+    if (updateInfoIsUpdated(pInfo->pUpdateInfo, pInfo->pRes->info.uid, ts[i])) {
+      taosArrayPush(pInfo->tsArray, ts+i);
+    }
+  }
+  if (taosArrayGetSize(pInfo->tsArray) > 0) {
+    //TODO(liuyao) get from tsdb
+    // SSDataBlock* p = createOneDataBlock(pInfo->pRes, true);
+    // p->info.type = STREAM_INVERT;
+    // taosArrayClear(pInfo->tsArray);
+    // return p;
+    return NULL;
+  }
+  return NULL;
+}
+
 static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
   // NOTE: this operator does never check if current status is done or not
   SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;
@@ -523,8 +542,8 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     return NULL;
   }
 
+  size_t total = taosArrayGetSize(pInfo->pBlockLists);
   if (pInfo->blockType == STREAM_DATA_TYPE_SSDATA_BLOCK) {
-    size_t total = taosArrayGetSize(pInfo->pBlockLists);
     if (pInfo->validBlockIndex >= total) {
       doClearBufferedBlocks(pInfo);
       pOperator->status = OP_EXEC_DONE;
@@ -534,6 +553,14 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     int32_t current = pInfo->validBlockIndex++;
     return taosArrayGetP(pInfo->pBlockLists, current);
   } else {
+    if (total > 0) {
+      ASSERT(total == 2);
+      SSDataBlock* pRes = taosArrayGetP(pInfo->pBlockLists, 0);
+      SSDataBlock* pUpRes = taosArrayGetP(pInfo->pBlockLists, 1);
+      blockDataDestroy(pUpRes);
+      taosArrayClear(pInfo->pBlockLists);
+      return pRes;
+    }
     SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
     blockDataCleanup(pInfo->pRes);
 
@@ -554,6 +581,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
       pInfo->pRes->info.groupId = groupId;
       pInfo->pRes->info.rows = numOfRows;
       pInfo->pRes->info.uid = uid;
+      pInfo->pRes->info.type = STREAM_NORMAL;
 
       int32_t numOfCols = pInfo->pRes->info.numOfCols;
       for (int32_t i = 0; i < numOfCols; ++i) {
@@ -598,6 +626,13 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
 
     if (rows == 0) {
       pOperator->status = OP_EXEC_DONE;
+    } else {
+      SSDataBlock* upRes = getUpdateDataBlock(pInfo);
+      if (upRes) {
+        taosArrayPush(pInfo->pBlockLists, &(pInfo->pRes));
+        taosArrayPush(pInfo->pBlockLists, &upRes);
+        return upRes;
+      }
     }
 
     return (rows == 0) ? NULL : pInfo->pRes;
@@ -636,6 +671,21 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
     goto _error;
   }
 
+  pInfo->tsArray = taosArrayInit(4, sizeof(TSKEY));
+  if (pInfo->tsArray == NULL) {
+    taosMemoryFreeClear(pInfo);
+    taosMemoryFreeClear(pOperator);
+    return NULL;
+  }
+
+  pInfo->primaryTsIndex = 0; //TODO(liuyao) get it from physical plan
+  pInfo->pUpdateInfo = updateInfoInit(60000, 0, 100); //TODO(liuyao) get it from physical plan
+  if (pInfo->pUpdateInfo == NULL) {
+    taosMemoryFreeClear(pInfo);
+    taosMemoryFreeClear(pOperator);
+    return NULL;
+  }
+
   pInfo->readerHandle = streamReadHandle;
   pInfo->pRes         = pResBlock;
   pInfo->pCondition   = pCondition;

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/function/inc/builtinsimpl.h

@@ -30,10 +30,12 @@ int32_t functionFinalizeWithResultBuf(SqlFunctionCtx* pCtx, SSDataBlock* pBlock,
 EFuncDataRequired countDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
 bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 int32_t countFunction(SqlFunctionCtx *pCtx);
+int32_t countInvertFunction(SqlFunctionCtx *pCtx);
 
 EFuncDataRequired statisDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
 bool getSumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 int32_t sumFunction(SqlFunctionCtx *pCtx);
+int32_t sumInvertFunction(SqlFunctionCtx *pCtx);
 
 bool minFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 bool maxFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
@@ -45,11 +47,13 @@ bool getAvgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool avgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t avgFunction(SqlFunctionCtx* pCtx);
 int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+int32_t avgInvertFunction(SqlFunctionCtx* pCtx);
 
 bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool stddevFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t stddevFunction(SqlFunctionCtx* pCtx);
 int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+int32_t stddevInvertFunction(SqlFunctionCtx* pCtx);
 
 bool getPercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool percentileFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);