Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/stream_compression

include/common/tmsg.h

@@ -866,7 +866,8 @@ int32_t tSerializeSDbCfgReq(void* buf, int32_t bufLen, SDbCfgReq* pReq);
 int32_t tDeserializeSDbCfgReq(void* buf, int32_t bufLen, SDbCfgReq* pReq);
 
 typedef struct {
-  char db[TSDB_DB_FNAME_LEN];
+  char    db[TSDB_DB_FNAME_LEN];
+  int32_t maxSpeed;
 } STrimDbReq;
 
 int32_t tSerializeSTrimDbReq(void* buf, int32_t bufLen, STrimDbReq* pReq);
@@ -1426,10 +1427,10 @@ typedef struct {
 
 typedef struct {
   SExplainRsp rsp;
-  uint64_t qId; 
+  uint64_t    qId;
-  uint64_t tId; 
+  uint64_t    tId;
-  int64_t rId; 
+  int64_t     rId;
-  int32_t eId;
+  int32_t     eId;
 } SExplainLocalRsp;
 
 typedef struct STableScanAnalyzeInfo {
@@ -1446,7 +1447,7 @@ typedef struct STableScanAnalyzeInfo {
 
 int32_t tSerializeSExplainRsp(void* buf, int32_t bufLen, SExplainRsp* pRsp);
 int32_t tDeserializeSExplainRsp(void* buf, int32_t bufLen, SExplainRsp* pRsp);
-void tFreeSExplainRsp(SExplainRsp *pRsp);
+void    tFreeSExplainRsp(SExplainRsp* pRsp);
 
 typedef struct {
   char    fqdn[TSDB_FQDN_LEN];  // end point, hostname:port
@@ -1729,6 +1730,8 @@ typedef struct {
   int64_t maxDelay;
   int64_t watermark;
   int8_t  igExpired;
+  int32_t numOfTags;
+  SArray* pTags;  // array of SField
 } SCMCreateStreamReq;
 
 typedef struct {

include/common/ttokendef.h

@@ -107,227 +107,229 @@
 #define TK_TABLE_PREFIX         89
 #define TK_TABLE_SUFFIX         90
 #define TK_NK_COLON             91
-#define TK_TABLE                92
+#define TK_MAX_SPEED            92
-#define TK_NK_LP                93
+#define TK_TABLE                93
-#define TK_NK_RP                94
+#define TK_NK_LP                94
-#define TK_STABLE               95
+#define TK_NK_RP                95
-#define TK_ADD                  96
+#define TK_STABLE               96
-#define TK_COLUMN               97
+#define TK_ADD                  97
-#define TK_MODIFY               98
+#define TK_COLUMN               98
-#define TK_RENAME               99
+#define TK_MODIFY               99
-#define TK_TAG                  100
+#define TK_RENAME               100
-#define TK_SET                  101
+#define TK_TAG                  101
-#define TK_NK_EQ                102
+#define TK_SET                  102
-#define TK_USING                103
+#define TK_NK_EQ                103
-#define TK_TAGS                 104
+#define TK_USING                104
-#define TK_COMMENT              105
+#define TK_TAGS                 105
-#define TK_BOOL                 106
+#define TK_COMMENT              106
-#define TK_TINYINT              107
+#define TK_BOOL                 107
-#define TK_SMALLINT             108
+#define TK_TINYINT              108
-#define TK_INT                  109
+#define TK_SMALLINT             109
-#define TK_INTEGER              110
+#define TK_INT                  110
-#define TK_BIGINT               111
+#define TK_INTEGER              111
-#define TK_FLOAT                112
+#define TK_BIGINT               112
-#define TK_DOUBLE               113
+#define TK_FLOAT                113
-#define TK_BINARY               114
+#define TK_DOUBLE               114
-#define TK_TIMESTAMP            115
+#define TK_BINARY               115
-#define TK_NCHAR                116
+#define TK_TIMESTAMP            116
-#define TK_UNSIGNED             117
+#define TK_NCHAR                117
-#define TK_JSON                 118
+#define TK_UNSIGNED             118
-#define TK_VARCHAR              119
+#define TK_JSON                 119
-#define TK_MEDIUMBLOB           120
+#define TK_VARCHAR              120
-#define TK_BLOB                 121
+#define TK_MEDIUMBLOB           121
-#define TK_VARBINARY            122
+#define TK_BLOB                 122
-#define TK_DECIMAL              123
+#define TK_VARBINARY            123
-#define TK_MAX_DELAY            124
+#define TK_DECIMAL              124
-#define TK_WATERMARK            125
+#define TK_MAX_DELAY            125
-#define TK_ROLLUP               126
+#define TK_WATERMARK            126
-#define TK_TTL                  127
+#define TK_ROLLUP               127
-#define TK_SMA                  128
+#define TK_TTL                  128
-#define TK_FIRST                129
+#define TK_SMA                  129
-#define TK_LAST                 130
+#define TK_FIRST                130
-#define TK_SHOW                 131
+#define TK_LAST                 131
-#define TK_DATABASES            132
+#define TK_SHOW                 132
-#define TK_TABLES               133
+#define TK_DATABASES            133
-#define TK_STABLES              134
+#define TK_TABLES               134
-#define TK_MNODES               135
+#define TK_STABLES              135
-#define TK_MODULES              136
+#define TK_MNODES               136
-#define TK_QNODES               137
+#define TK_MODULES              137
-#define TK_FUNCTIONS            138
+#define TK_QNODES               138
-#define TK_INDEXES              139
+#define TK_FUNCTIONS            139
-#define TK_ACCOUNTS             140
+#define TK_INDEXES              140
-#define TK_APPS                 141
+#define TK_ACCOUNTS             141
-#define TK_CONNECTIONS          142
+#define TK_APPS                 142
-#define TK_LICENCES             143
+#define TK_CONNECTIONS          143
-#define TK_GRANTS               144
+#define TK_LICENCES             144
-#define TK_QUERIES              145
+#define TK_GRANTS               145
-#define TK_SCORES               146
+#define TK_QUERIES              146
-#define TK_TOPICS               147
+#define TK_SCORES               147
-#define TK_VARIABLES            148
+#define TK_TOPICS               148
-#define TK_BNODES               149
+#define TK_VARIABLES            149
-#define TK_SNODES               150
+#define TK_BNODES               150
-#define TK_CLUSTER              151
+#define TK_SNODES               151
-#define TK_TRANSACTIONS         152
+#define TK_CLUSTER              152
-#define TK_DISTRIBUTED          153
+#define TK_TRANSACTIONS         153
-#define TK_CONSUMERS            154
+#define TK_DISTRIBUTED          154
-#define TK_SUBSCRIPTIONS        155
+#define TK_CONSUMERS            155
-#define TK_VNODES               156
+#define TK_SUBSCRIPTIONS        156
-#define TK_LIKE                 157
+#define TK_VNODES               157
-#define TK_INDEX                158
+#define TK_LIKE                 158
-#define TK_FUNCTION             159
+#define TK_INDEX                159
-#define TK_INTERVAL             160
+#define TK_FUNCTION             160
-#define TK_TOPIC                161
+#define TK_INTERVAL             161
-#define TK_AS                   162
+#define TK_TOPIC                162
-#define TK_WITH                 163
+#define TK_AS                   163
-#define TK_META                 164
+#define TK_WITH                 164
-#define TK_CONSUMER             165
+#define TK_META                 165
-#define TK_GROUP                166
+#define TK_CONSUMER             166
-#define TK_DESC                 167
+#define TK_GROUP                167
-#define TK_DESCRIBE             168
+#define TK_DESC                 168
-#define TK_RESET                169
+#define TK_DESCRIBE             169
-#define TK_QUERY                170
+#define TK_RESET                170
-#define TK_CACHE                171
+#define TK_QUERY                171
-#define TK_EXPLAIN              172
+#define TK_CACHE                172
-#define TK_ANALYZE              173
+#define TK_EXPLAIN              173
-#define TK_VERBOSE              174
+#define TK_ANALYZE              174
-#define TK_NK_BOOL              175
+#define TK_VERBOSE              175
-#define TK_RATIO                176
+#define TK_NK_BOOL              176
-#define TK_NK_FLOAT             177
+#define TK_RATIO                177
-#define TK_OUTPUTTYPE           178
+#define TK_NK_FLOAT             178
-#define TK_AGGREGATE            179
+#define TK_OUTPUTTYPE           179
-#define TK_BUFSIZE              180
+#define TK_AGGREGATE            180
-#define TK_STREAM               181
+#define TK_BUFSIZE              181
-#define TK_INTO                 182
+#define TK_STREAM               182
-#define TK_TRIGGER              183
+#define TK_INTO                 183
-#define TK_AT_ONCE              184
+#define TK_TRIGGER              184
-#define TK_WINDOW_CLOSE         185
+#define TK_AT_ONCE              185
-#define TK_IGNORE               186
+#define TK_WINDOW_CLOSE         186
-#define TK_EXPIRED              187
+#define TK_IGNORE               187
-#define TK_KILL                 188
+#define TK_EXPIRED              188
-#define TK_CONNECTION           189
+#define TK_SUBTABLE             189
-#define TK_TRANSACTION          190
+#define TK_KILL                 190
-#define TK_BALANCE              191
+#define TK_CONNECTION           191
-#define TK_VGROUP               192
+#define TK_TRANSACTION          192
-#define TK_MERGE                193
+#define TK_BALANCE              193
-#define TK_REDISTRIBUTE         194
+#define TK_VGROUP               194
-#define TK_SPLIT                195
+#define TK_MERGE                195
-#define TK_DELETE               196
+#define TK_REDISTRIBUTE         196
-#define TK_INSERT               197
+#define TK_SPLIT                197
-#define TK_NULL                 198
+#define TK_DELETE               198
-#define TK_NK_QUESTION          199
+#define TK_INSERT               199
-#define TK_NK_ARROW             200
+#define TK_NULL                 200
-#define TK_ROWTS                201
+#define TK_NK_QUESTION          201
-#define TK_TBNAME               202
+#define TK_NK_ARROW             202
-#define TK_QSTART               203
+#define TK_ROWTS                203
-#define TK_QEND                 204
+#define TK_TBNAME               204
-#define TK_QDURATION            205
+#define TK_QSTART               205
-#define TK_WSTART               206
+#define TK_QEND                 206
-#define TK_WEND                 207
+#define TK_QDURATION            207
-#define TK_WDURATION            208
+#define TK_WSTART               208
-#define TK_CAST                 209
+#define TK_WEND                 209
-#define TK_NOW                  210
+#define TK_WDURATION            210
-#define TK_TODAY                211
+#define TK_CAST                 211
-#define TK_TIMEZONE             212
+#define TK_NOW                  212
-#define TK_CLIENT_VERSION       213
+#define TK_TODAY                213
-#define TK_SERVER_VERSION       214
+#define TK_TIMEZONE             214
-#define TK_SERVER_STATUS        215
+#define TK_CLIENT_VERSION       215
-#define TK_CURRENT_USER         216
+#define TK_SERVER_VERSION       216
-#define TK_COUNT                217
+#define TK_SERVER_STATUS        217
-#define TK_LAST_ROW             218
+#define TK_CURRENT_USER         218
-#define TK_CASE                 219
+#define TK_COUNT                219
-#define TK_END                  220
+#define TK_LAST_ROW             220
-#define TK_WHEN                 221
+#define TK_CASE                 221
-#define TK_THEN                 222
+#define TK_END                  222
-#define TK_ELSE                 223
+#define TK_WHEN                 223
-#define TK_BETWEEN              224
+#define TK_THEN                 224
-#define TK_IS                   225
+#define TK_ELSE                 225
-#define TK_NK_LT                226
+#define TK_BETWEEN              226
-#define TK_NK_GT                227
+#define TK_IS                   227
-#define TK_NK_LE                228
+#define TK_NK_LT                228
-#define TK_NK_GE                229
+#define TK_NK_GT                229
-#define TK_NK_NE                230
+#define TK_NK_LE                230
-#define TK_MATCH                231
+#define TK_NK_GE                231
-#define TK_NMATCH               232
+#define TK_NK_NE                232
-#define TK_CONTAINS             233
+#define TK_MATCH                233
-#define TK_IN                   234
+#define TK_NMATCH               234
-#define TK_JOIN                 235
+#define TK_CONTAINS             235
-#define TK_INNER                236
+#define TK_IN                   236
-#define TK_SELECT               237
+#define TK_JOIN                 237
-#define TK_DISTINCT             238
+#define TK_INNER                238
-#define TK_WHERE                239
+#define TK_SELECT               239
-#define TK_PARTITION            240
+#define TK_DISTINCT             240
-#define TK_BY                   241
+#define TK_WHERE                241
-#define TK_SESSION              242
+#define TK_PARTITION            242
-#define TK_STATE_WINDOW         243
+#define TK_BY                   243
-#define TK_SLIDING              244
+#define TK_SESSION              244
-#define TK_FILL                 245
+#define TK_STATE_WINDOW         245
-#define TK_VALUE                246
+#define TK_SLIDING              246
-#define TK_NONE                 247
+#define TK_FILL                 247
-#define TK_PREV                 248
+#define TK_VALUE                248
-#define TK_LINEAR               249
+#define TK_NONE                 249
-#define TK_NEXT                 250
+#define TK_PREV                 250
-#define TK_HAVING               251
+#define TK_LINEAR               251
-#define TK_RANGE                252
+#define TK_NEXT                 252
-#define TK_EVERY                253
+#define TK_HAVING               253
-#define TK_ORDER                254
+#define TK_RANGE                254
-#define TK_SLIMIT               255
+#define TK_EVERY                255
-#define TK_SOFFSET              256
+#define TK_ORDER                256
-#define TK_LIMIT                257
+#define TK_SLIMIT               257
-#define TK_OFFSET               258
+#define TK_SOFFSET              258
-#define TK_ASC                  259
+#define TK_LIMIT                259
-#define TK_NULLS                260
+#define TK_OFFSET               260
-#define TK_ABORT                261
+#define TK_ASC                  261
-#define TK_AFTER                262
+#define TK_NULLS                262
-#define TK_ATTACH               263
+#define TK_ABORT                263
-#define TK_BEFORE               264
+#define TK_AFTER                264
-#define TK_BEGIN                265
+#define TK_ATTACH               265
-#define TK_BITAND               266
+#define TK_BEFORE               266
-#define TK_BITNOT               267
+#define TK_BEGIN                267
-#define TK_BITOR                268
+#define TK_BITAND               268
-#define TK_BLOCKS               269
+#define TK_BITNOT               269
-#define TK_CHANGE               270
+#define TK_BITOR                270
-#define TK_COMMA                271
+#define TK_BLOCKS               271
-#define TK_COMPACT              272
+#define TK_CHANGE               272
-#define TK_CONCAT               273
+#define TK_COMMA                273
-#define TK_CONFLICT             274
+#define TK_COMPACT              274
-#define TK_COPY                 275
+#define TK_CONCAT               275
-#define TK_DEFERRED             276
+#define TK_CONFLICT             276
-#define TK_DELIMITERS           277
+#define TK_COPY                 277
-#define TK_DETACH               278
+#define TK_DEFERRED             278
-#define TK_DIVIDE               279
+#define TK_DELIMITERS           279
-#define TK_DOT                  280
+#define TK_DETACH               280
-#define TK_EACH                 281
+#define TK_DIVIDE               281
-#define TK_FAIL                 282
+#define TK_DOT                  282
-#define TK_FILE                 283
+#define TK_EACH                 283
-#define TK_FOR                  284
+#define TK_FAIL                 284
-#define TK_GLOB                 285
+#define TK_FILE                 285
-#define TK_ID                   286
+#define TK_FOR                  286
-#define TK_IMMEDIATE            287
+#define TK_GLOB                 287
-#define TK_IMPORT               288
+#define TK_ID                   288
-#define TK_INITIALLY            289
+#define TK_IMMEDIATE            289
-#define TK_INSTEAD              290
+#define TK_IMPORT               290
-#define TK_ISNULL               291
+#define TK_INITIALLY            291
-#define TK_KEY                  292
+#define TK_INSTEAD              292
-#define TK_NK_BITNOT            293
+#define TK_ISNULL               293
-#define TK_NK_SEMI              294
+#define TK_KEY                  294
-#define TK_NOTNULL              295
+#define TK_NK_BITNOT            295
-#define TK_OF                   296
+#define TK_NK_SEMI              296
-#define TK_PLUS                 297
+#define TK_NOTNULL              297
-#define TK_PRIVILEGE            298
+#define TK_OF                   298
-#define TK_RAISE                299
+#define TK_PLUS                 299
-#define TK_REPLACE              300
+#define TK_PRIVILEGE            300
-#define TK_RESTRICT             301
+#define TK_RAISE                301
-#define TK_ROW                  302
+#define TK_REPLACE              302
-#define TK_SEMI                 303
+#define TK_RESTRICT             303
-#define TK_STAR                 304
+#define TK_ROW                  304
-#define TK_STATEMENT            305
+#define TK_SEMI                 305
-#define TK_STRING               306
+#define TK_STAR                 306
-#define TK_TIMES                307
+#define TK_STATEMENT            307
-#define TK_UPDATE               308
+#define TK_STRING               308
-#define TK_VALUES               309
+#define TK_TIMES                309
-#define TK_VARIABLE             310
+#define TK_UPDATE               310
-#define TK_VIEW                 311
+#define TK_VALUES               311
-#define TK_WAL                  312
+#define TK_VARIABLE             312
+#define TK_VIEW                 313
+#define TK_WAL                  314
 
 #define TK_NK_SPACE   300
 #define TK_NK_COMMENT 301

include/libs/nodes/cmdnodes.h

@@ -119,6 +119,7 @@ typedef struct SFlushDatabaseStmt {
 typedef struct STrimDatabaseStmt {
   ENodeType type;
   char      dbName[TSDB_DB_NAME_LEN];
+  int32_t   maxSpeed;
 } STrimDatabaseStmt;
 
 typedef struct STableOptions {
@@ -383,6 +384,8 @@ typedef struct SCreateStreamStmt {
   bool            ignoreExists;
   SStreamOptions* pOptions;
   SNode*          pQuery;
+  SNodeList*      pTags;
+  SNode*          pSubtable;
 } SCreateStreamStmt;
 
 typedef struct SDropStreamStmt {

include/libs/nodes/nodes.h

@@ -27,9 +27,9 @@ extern "C" {
 
 #define LIST_LENGTH(l) (NULL != (l) ? (l)->length : 0)
 
-#define FOREACH(node, list)                                       \
+#define FOREACH(node, list)                                               \
-  for (SListCell* cell = (NULL != (list) ? (list)->pHead : NULL); \
+  for (SListCell *cell = (NULL != (list) ? (list)->pHead : NULL), *pNext; \
-       (NULL != cell ? (node = cell->pNode, true) : (node = NULL, false)); cell = cell->pNext)
+       (NULL != cell ? (node = cell->pNode, pNext = cell->pNext, true) : (node = NULL, pNext = NULL, false)); cell = pNext)
 
 #define REPLACE_NODE(newNode) cell->pNode = (SNode*)(newNode)
 

include/libs/nodes/plannodes.h

@@ -94,6 +94,8 @@ typedef struct SScanLogicNode {
   SArray*       pSmaIndexes;
   SNodeList*    pGroupTags;
   bool          groupSort;
+  SNodeList*    pTags;      // for create stream
+  SNode*        pSubtable;  // for create stream
   int8_t        cacheLastMode;
   bool          hasNormalCols;  // neither tag column nor primary key tag column
   bool          sortPrimaryKey;
@@ -233,6 +235,8 @@ typedef struct SSortLogicNode {
 typedef struct SPartitionLogicNode {
   SLogicNode node;
   SNodeList* pPartitionKeys;
+  SNodeList* pTags;
+  SNode*     pSubtable;
 } SPartitionLogicNode;
 
 typedef enum ESubplanType {
@@ -332,6 +336,8 @@ typedef struct STableScanPhysiNode {
   SNodeList*     pDynamicScanFuncs;
   SNodeList*     pGroupTags;
   bool           groupSort;
+  SNodeList*     pTags;
+  SNode*         pSubtable;
   int64_t        interval;
   int64_t        offset;
   int64_t        sliding;
@@ -495,7 +501,11 @@ typedef struct SPartitionPhysiNode {
   SNodeList* pTargets;
 } SPartitionPhysiNode;
 
-typedef SPartitionPhysiNode SStreamPartitionPhysiNode;
+typedef struct SStreamPartitionPhysiNode {
+  SPartitionPhysiNode part;
+  SNodeList*          pTags;
+  SNode*              pSubtable;
+} SStreamPartitionPhysiNode;
 
 typedef struct SDataSinkNode {
   ENodeType           type;

include/libs/nodes/querynodes.h

@@ -261,6 +261,8 @@ typedef struct SSelectStmt {
   SNode*      pFromTable;
   SNode*      pWhere;
   SNodeList*  pPartitionByList;
+  SNodeList*  pTags;      // for create stream
+  SNode*      pSubtable;  // for create stream
   SNode*      pWindow;
   SNodeList*  pGroupByList;  // SGroupingSetNode
   SNode*      pHaving;

include/os/osSocket.h

@@ -167,7 +167,7 @@ uint32_t    ip2uint(const char *const ip_addr);
 void        taosIgnSIGPIPE();
 void        taosSetMaskSIGPIPE();
 uint32_t    taosInetAddr(const char *ipAddr);
-const char *taosInetNtoa(struct in_addr ipInt);
+const char *taosInetNtoa(struct in_addr ipInt, char *dstStr, int32_t len);
 
 #ifdef __cplusplus
 }

include/os/osString.h

@@ -77,7 +77,6 @@ int32_t taosWcharsWidth(TdWchar *pWchar, int32_t size);
 int32_t taosMbToWchar(TdWchar *pWchar, const char *pStr, int32_t size);
 int32_t taosMbsToWchars(TdWchar *pWchars, const char *pStrs, int32_t size);
 int32_t taosWcharToMb(char *pStr, TdWchar wchar);
-int32_t taosWcharsToMbs(char *pStrs, TdWchar *pWchars, int32_t size);
 
 char *taosStrCaseStr(const char *str, const char *pattern);
 

source/client/src/clientSml.c

@@ -79,7 +79,7 @@
 #define NCHAR_ADD_LEN  3  // L"nchar"   3 means L" "
 
 #define MAX_RETRY_TIMES 5
-#define LINE_BATCH      20000
+#define LINE_BATCH      2000
 //=================================================================================================
 typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;
 
@@ -1532,6 +1532,7 @@ static SSmlHandle* smlBuildSmlInfo(STscObj* pTscObj, SRequestObj* request, SMLPr
     info->pRequest = request;
     info->msgBuf.buf = info->pRequest->msgBuf;
     info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
+    info->pRequest->stmtType = info->pQuery->pRoot->type;
   }
 
   info->exec = smlInitHandle(info->pQuery);
@@ -2332,6 +2333,9 @@ static int32_t smlInsertData(SSmlHandle *info) {
   //  info->affectedRows = taos_affected_rows(info->pRequest);
   //  return info->pRequest->code;
 
+  SAppClusterSummary *pActivity = &info->taos->pAppInfo->summary;
+  atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);
+
   launchAsyncQuery(info->pRequest, info->pQuery, NULL);
   return TSDB_CODE_SUCCESS;
 }

source/common/src/tglobal.c

@@ -685,7 +685,9 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsQueryBufferSize = cfgGetItem(pCfg, "queryBufferSize")->i32;
   tsPrintAuth = cfgGetItem(pCfg, "printAuth")->bval;
 
+#if !defined(WINDOWS) && !defined(DARWIN)
   tsMultiProcess = cfgGetItem(pCfg, "multiProcess")->bval;
+#endif
   tsMnodeShmSize = cfgGetItem(pCfg, "mnodeShmSize")->i32;
   tsVnodeShmSize = cfgGetItem(pCfg, "vnodeShmSize")->i32;
   tsQnodeShmSize = cfgGetItem(pCfg, "qnodeShmSize")->i32;
@@ -919,7 +921,9 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
         }
         case 'u': {
           if (strcasecmp("multiProcess", name) == 0) {
+          #if !defined(WINDOWS) && !defined(DARWIN)
             tsMultiProcess = cfgGetItem(pCfg, "multiProcess")->bval;
+          #endif
           } else if (strcasecmp("udfDebugFlag", name) == 0) {
             udfDebugFlag = cfgGetItem(pCfg, "udfDebugFlag")->i32;
           }

source/common/src/tmsg.c

@@ -2682,6 +2682,7 @@ int32_t tSerializeSTrimDbReq(void *buf, int32_t bufLen, STrimDbReq *pReq) {
 
   if (tStartEncode(&encoder) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->db) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->maxSpeed) < 0) return -1;
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -2695,6 +2696,7 @@ int32_t tDeserializeSTrimDbReq(void *buf, int32_t bufLen, STrimDbReq *pReq) {
 
   if (tStartDecode(&decoder) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->maxSpeed) < 0) return -1;
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);
@@ -4373,8 +4375,7 @@ int32_t tDeserializeSExplainRsp(void *buf, int32_t bufLen, SExplainRsp *pRsp) {
     if (tDecodeDouble(&decoder, &pRsp->subplanInfo[i].totalCost) < 0) return -1;
     if (tDecodeU64(&decoder, &pRsp->subplanInfo[i].numOfRows) < 0) return -1;
     if (tDecodeU32(&decoder, &pRsp->subplanInfo[i].verboseLen) < 0) return -1;
-    if (tDecodeBinaryAlloc(&decoder, &pRsp->subplanInfo[i].verboseInfo, NULL) < 0)
+    if (tDecodeBinaryAlloc(&decoder, &pRsp->subplanInfo[i].verboseInfo, NULL) < 0) return -1;
-      return -1;
   }
 
   tEndDecode(&decoder);
@@ -4826,6 +4827,14 @@ int32_t tSerializeSCMCreateStreamReq(void *buf, int32_t bufLen, const SCMCreateS
   if (tEncodeI8(&encoder, pReq->igExpired) < 0) return -1;
   if (sqlLen > 0 && tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
   if (astLen > 0 && tEncodeCStr(&encoder, pReq->ast) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->numOfTags) < 0) return -1;
+  for (int32_t i = 0; i < pReq->numOfTags; ++i) {
+    SField *pField = taosArrayGet(pReq->pTags, i);
+    if (tEncodeI8(&encoder, pField->type) < 0) return -1;
+    if (tEncodeI8(&encoder, pField->flags) < 0) return -1;
+    if (tEncodeI32(&encoder, pField->bytes) < 0) return -1;
+    if (tEncodeCStr(&encoder, pField->name) < 0) return -1;
+  }
 
   tEndEncode(&encoder);
 
@@ -4864,6 +4873,28 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
     if (pReq->ast == NULL) return -1;
     if (tDecodeCStrTo(&decoder, pReq->ast) < 0) return -1;
   }
+
+  if (tDecodeI32(&decoder, &pReq->numOfTags) < 0) return -1;
+  if (pReq->numOfTags > 0) {
+    pReq->pTags = taosArrayInit(pReq->numOfTags, sizeof(SField));
+    if (pReq->pTags == NULL) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return -1;
+    }
+
+    for (int32_t i = 0; i < pReq->numOfTags; ++i) {
+      SField field = {0};
+      if (tDecodeI8(&decoder, &field.type) < 0) return -1;
+      if (tDecodeI8(&decoder, &field.flags) < 0) return -1;
+      if (tDecodeI32(&decoder, &field.bytes) < 0) return -1;
+      if (tDecodeCStrTo(&decoder, field.name) < 0) return -1;
+      if (taosArrayPush(pReq->pTags, &field) == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return -1;
+      }
+    }
+  }
+
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);

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

@@ -1604,7 +1604,7 @@ static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)dbname, false);
 
-    char stbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
+    char stbname[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
     STR_WITH_MAXSIZE_TO_VARSTR(stbname, mndGetDbStr(pTrans->stbname), pShow->pMeta->pSchemas[cols].bytes);
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)stbname, false);

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

@@ -97,9 +97,8 @@ typedef struct SDiskDataBuilder SDiskDataBuilder;
 static FORCE_INLINE int64_t tsdbLogicToFileSize(int64_t lSize, int32_t szPage) {
   int64_t fOffSet = LOGIC_TO_FILE_OFFSET(lSize, szPage);
   int64_t pgno = OFFSET_PGNO(fOffSet, szPage);
-  int32_t szPageCont = PAGE_CONTENT_SIZE(szPage);
 
-  if (fOffSet % szPageCont == 0) {
+  if (fOffSet % szPage == 0) {
     pgno--;
   }
 

source/libs/executor/inc/executorimpl.h

@@ -990,7 +990,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi
 SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode,
                                            SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode,
+SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode,
                                                  SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);

source/libs/executor/src/executorimpl.c

@@ -1110,7 +1110,8 @@ void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numO
   }
 }
 
-static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep, int32_t status);
+static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep,
+                                                int32_t status);
 
 void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, const SArray* pColMatchInfo) {
   if (pFilterNode == NULL || pBlock->info.rows == 0) {
@@ -1120,12 +1121,12 @@ void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, const SArray* pColM
   SFilterInfo* filter = NULL;
 
   // todo move to the initialization function
-  int32_t code = filterInitFromNode((SNode*)pFilterNode, &filter, 0);
+  int32_t            code = filterInitFromNode((SNode*)pFilterNode, &filter, 0);
   SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};
   code = filterSetDataFromSlotId(filter, &param1);
 
   SColumnInfoData* p = NULL;
-  int32_t status = 0;
+  int32_t          status = 0;
 
   // todo the keep seems never to be True??
   bool keep = filterExecute(filter, pBlock, &p, NULL, param1.numOfCols, &status);
@@ -1201,7 +1202,7 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoD
   }
 }
 
-  void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId) {
+void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId) {
   // for simple group by query without interval, all the tables belong to one group result.
   SExecTaskInfo*    pTaskInfo = pOperator->pTaskInfo;
   SAggOperatorInfo* pAggInfo = pOperator->info;
@@ -1787,7 +1788,9 @@ static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInf
 
   if (pSource->localExec) {
     SDataBuf pBuf = {0};
-    int32_t code = (*pTaskInfo->localFetch.fp)(pTaskInfo->localFetch.handle, pSource->schedId, pTaskInfo->id.queryId, pSource->taskId, 0, pSource->execId, &pBuf.pData, pTaskInfo->localFetch.explainRes);
+    int32_t  code =
+        (*pTaskInfo->localFetch.fp)(pTaskInfo->localFetch.handle, pSource->schedId, pTaskInfo->id.queryId,
+                                    pSource->taskId, 0, pSource->execId, &pBuf.pData, pTaskInfo->localFetch.explainRes);
     loadRemoteDataCallback(pWrapper, &pBuf, code);
     taosMemoryFree(pWrapper);
   } else {
@@ -1798,8 +1801,8 @@ static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInf
     }
 
     qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %d/%" PRIzu,
-           GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId, pSource->execId,
+           GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId,
-           sourceIndex, totalSources);
+           pSource->execId, sourceIndex, totalSources);
 
     pMsg->header.vgId = htonl(pSource->addr.nodeId);
     pMsg->sId = htobe64(pSource->schedId);
@@ -1824,7 +1827,8 @@ static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInf
     pMsgSendInfo->fp = loadRemoteDataCallback;
 
     int64_t transporterId = 0;
-    int32_t code = asyncSendMsgToServer(pExchangeInfo->pTransporter, &pSource->addr.epSet, &transporterId, pMsgSendInfo);
+    int32_t code =
+        asyncSendMsgToServer(pExchangeInfo->pTransporter, &pSource->addr.epSet, &transporterId, pMsgSendInfo);
   }
 
   return TSDB_CODE_SUCCESS;
@@ -3356,9 +3360,7 @@ static void cleanupTableSchemaInfo(SSchemaInfo* pSchemaInfo) {
   tDeleteSSchemaWrapper(pSchemaInfo->qsw);
 }
 
-static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) {
+static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) { tDeleteSSchemaWrapper(pStreamInfo->schema); }
-  tDeleteSSchemaWrapper(pStreamInfo->schema);
-}
 
 static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
   taosArrayClear(pTableListInfo->pGroupList);
@@ -3539,7 +3541,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
       STableScanInfo* pScanInfo = pOperator->info;
       pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
     } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
-      pOperator = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode, pTaskInfo);
+      pOperator = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode,
+                                             pTaskInfo);
     } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
       STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
       if (pHandle->vnode) {
@@ -3729,7 +3732,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
   } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
     pOptr = createPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION == type) {
-    pOptr = createStreamPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo);
+    pOptr = createStreamPartitionOperatorInfo(ops[0], (SStreamPartitionPhysiNode*)pPhyNode, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE == type) {
     SStateWinodwPhysiNode* pStateNode = (SStateWinodwPhysiNode*)pPhyNode;
     pOptr = createStatewindowOperatorInfo(ops[0], pStateNode, pTaskInfo);

source/libs/executor/src/groupoperator.c

@@ -989,11 +989,11 @@ SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStr
     goto _error;
   }
   int32_t code = TSDB_CODE_SUCCESS;
-  pInfo->partitionSup.pGroupCols = extractPartitionColInfo(pPartNode->pPartitionKeys);
+  pInfo->partitionSup.pGroupCols = extractPartitionColInfo(pPartNode->part.pPartitionKeys);
 
-  if (pPartNode->pExprs != NULL) {
+  if (pPartNode->part.pExprs != NULL) {
     int32_t    num = 0;
-    SExprInfo* pCalExprInfo = createExprInfo(pPartNode->pExprs, NULL, &num);
+    SExprInfo* pCalExprInfo = createExprInfo(pPartNode->part.pExprs, NULL, &num);
     code = initExprSupp(&pInfo->scalarSup, pCalExprInfo, num);
     if (code != TSDB_CODE_SUCCESS) {
       goto _error;
@@ -1008,7 +1008,7 @@ SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStr
   }
   pInfo->partitionSup.needCalc = true;
 
-  SSDataBlock* pResBlock = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);
+  SSDataBlock* pResBlock = createResDataBlock(pPartNode->part.node.pOutputDataBlockDesc);
   if (!pResBlock) {
     goto _error;
   }
@@ -1022,7 +1022,7 @@ SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStr
   pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT);
 
   int32_t    numOfCols = 0;
-  SExprInfo* pExprInfo = createExprInfo(pPartNode->pTargets, NULL, &numOfCols);
+  SExprInfo* pExprInfo = createExprInfo(pPartNode->part.pTargets, NULL, &numOfCols);
 
   pOperator->name = "StreamPartitionOperator";
   pOperator->blocking = false;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -381,6 +381,8 @@ static int32_t logicScanCopy(const SScanLogicNode* pSrc, SScanLogicNode* pDst) {
   COPY_SCALAR_FIELD(igExpired);
   CLONE_NODE_LIST_FIELD(pGroupTags);
   COPY_SCALAR_FIELD(groupSort);
+  CLONE_NODE_LIST_FIELD(pTags);
+  CLONE_NODE_FIELD(pSubtable);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -488,6 +490,8 @@ static int32_t logicSortCopy(const SSortLogicNode* pSrc, SSortLogicNode* pDst) {
 static int32_t logicPartitionCopy(const SPartitionLogicNode* pSrc, SPartitionLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
   CLONE_NODE_LIST_FIELD(pPartitionKeys);
+  CLONE_NODE_LIST_FIELD(pTags);
+  CLONE_NODE_FIELD(pSubtable);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/nodes/src/nodesCodeFuncs.c

@@ -1538,6 +1538,8 @@ static const char* jkTableScanPhysiPlanWatermark = "Watermark";
 static const char* jkTableScanPhysiPlanIgnoreExpired = "IgnoreExpired";
 static const char* jkTableScanPhysiPlanGroupTags = "GroupTags";
 static const char* jkTableScanPhysiPlanGroupSort = "GroupSort";
+static const char* jkTableScanPhysiPlanTags = "Tags";
+static const char* jkTableScanPhysiPlanSubtable = "Subtable";
 static const char* jkTableScanPhysiPlanAssignBlockUid = "AssignBlockUid";
 
 static int32_t physiTableScanNodeToJson(const void* pObj, SJson* pJson) {
@@ -1595,6 +1597,12 @@ static int32_t physiTableScanNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddBoolToObject(pJson, jkTableScanPhysiPlanGroupSort, pNode->groupSort);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodeListToJson(pJson, jkTableScanPhysiPlanTags, pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkTableScanPhysiPlanSubtable, nodeToJson, pNode->pSubtable);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddBoolToObject(pJson, jkTableScanPhysiPlanAssignBlockUid, pNode->assignBlockUid);
   }
@@ -1657,6 +1665,12 @@ static int32_t jsonToPhysiTableScanNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetBoolValue(pJson, jkTableScanPhysiPlanGroupSort, &pNode->groupSort);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeList(pJson, jkTableScanPhysiPlanTags, &pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkTableScanPhysiPlanSubtable, &pNode->pSubtable);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetBoolValue(pJson, jkTableScanPhysiPlanAssignBlockUid, &pNode->assignBlockUid);
   }
@@ -2270,6 +2284,37 @@ static int32_t jsonToPhysiPartitionNode(const SJson* pJson, void* pObj) {
   return code;
 }
 
+static const char* jkStreamPartitionPhysiPlanTags = "Tags";
+static const char* jkStreamPartitionPhysiPlanSubtable = "Subtable";
+
+static int32_t physiStreamPartitionNodeToJson(const void* pObj, SJson* pJson) {
+  const SStreamPartitionPhysiNode* pNode = (const SStreamPartitionPhysiNode*)pObj;
+
+  int32_t code = physiPartitionNodeToJson(pObj, pJson);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodeListToJson(pJson, jkStreamPartitionPhysiPlanTags, pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkStreamPartitionPhysiPlanSubtable, nodeToJson, pNode->pSubtable);
+  }
+
+  return code;
+}
+
+static int32_t jsonToPhysiStreamPartitionNode(const SJson* pJson, void* pObj) {
+  SStreamPartitionPhysiNode* pNode = (SStreamPartitionPhysiNode*)pObj;
+
+  int32_t code = jsonToPhysiPartitionNode(pJson, pObj);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeList(pJson, jkStreamPartitionPhysiPlanTags, &pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkStreamPartitionPhysiPlanSubtable, &pNode->pSubtable);
+  }
+
+  return code;
+}
+
 static const char* jkIndefRowsFuncPhysiPlanExprs = "Exprs";
 static const char* jkIndefRowsFuncPhysiPlanFuncs = "Funcs";
 
@@ -4109,6 +4154,8 @@ static const char* jkSelectStmtProjections = "Projections";
 static const char* jkSelectStmtFrom = "From";
 static const char* jkSelectStmtWhere = "Where";
 static const char* jkSelectStmtPartitionBy = "PartitionBy";
+static const char* jkSelectStmtTags = "Tags";
+static const char* jkSelectStmtSubtable = "Subtable";
 static const char* jkSelectStmtWindow = "Window";
 static const char* jkSelectStmtGroupBy = "GroupBy";
 static const char* jkSelectStmtHaving = "Having";
@@ -4134,6 +4181,12 @@ static int32_t selectStmtToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = nodeListToJson(pJson, jkSelectStmtPartitionBy, pNode->pPartitionByList);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodeListToJson(pJson, jkSelectStmtTags, pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddObject(pJson, jkSelectStmtSubtable, nodeToJson, pNode->pSubtable);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkSelectStmtWindow, nodeToJson, pNode->pWindow);
   }
@@ -4178,6 +4231,12 @@ static int32_t jsonToSelectStmt(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeList(pJson, jkSelectStmtPartitionBy, &pNode->pPartitionByList);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeList(pJson, jkSelectStmtTags, &pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = jsonToNodeObject(pJson, jkSelectStmtSubtable, &pNode->pSubtable);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = jsonToNodeObject(pJson, jkSelectStmtWindow, &pNode->pWindow);
   }
@@ -4586,8 +4645,9 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
     case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:
       return physiStateWindowNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
-    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
       return physiPartitionNodeToJson(pObj, pJson);
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
+      return physiStreamPartitionNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC:
       return physiIndefRowsFuncNodeToJson(pObj, pJson);
     case QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC:
@@ -4738,8 +4798,9 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
     case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:
       return jsonToPhysiStateWindowNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
-    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
       return jsonToPhysiPartitionNode(pJson, pObj);
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
+      return jsonToPhysiStreamPartitionNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC:
       return jsonToPhysiIndefRowsFuncNode(pJson, pObj);
     case QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC:

source/libs/nodes/src/nodesMsgFuncs.c

@@ -1993,7 +1993,9 @@ enum {
   PHY_TABLE_SCAN_CODE_SCAN = 1,
   PHY_TABLE_SCAN_CODE_INLINE_ATTRS,
   PHY_TABLE_SCAN_CODE_DYN_SCAN_FUNCS,
-  PHY_TABLE_SCAN_CODE_GROUP_TAGS
+  PHY_TABLE_SCAN_CODE_GROUP_TAGS,
+  PHY_TABLE_SCAN_CODE_TAGS,
+  PHY_TABLE_SCAN_CODE_SUBTABLE
 };
 
 static int32_t physiTableScanNodeInlineToMsg(const void* pObj, STlvEncoder* pEncoder) {
@@ -2062,6 +2064,12 @@ static int32_t physiTableScanNodeToMsg(const void* pObj, STlvEncoder* pEncoder)
   if (TSDB_CODE_SUCCESS == code) {
     code = tlvEncodeObj(pEncoder, PHY_TABLE_SCAN_CODE_GROUP_TAGS, nodeListToMsg, pNode->pGroupTags);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, PHY_TABLE_SCAN_CODE_TAGS, nodeListToMsg, pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, PHY_TABLE_SCAN_CODE_SUBTABLE, nodeToMsg, pNode->pSubtable);
+  }
 
   return code;
 }
@@ -2138,6 +2146,12 @@ static int32_t msgToPhysiTableScanNode(STlvDecoder* pDecoder, void* pObj) {
       case PHY_TABLE_SCAN_CODE_GROUP_TAGS:
         code = msgToNodeListFromTlv(pTlv, (void**)&pNode->pGroupTags);
         break;
+      case PHY_TABLE_SCAN_CODE_TAGS:
+        code = msgToNodeListFromTlv(pTlv, (void**)&pNode->pTags);
+        break;
+      case PHY_TABLE_SCAN_CODE_SUBTABLE:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pSubtable);
+        break;
       default:
         break;
     }
@@ -2914,6 +2928,46 @@ static int32_t msgToPhysiPartitionNode(STlvDecoder* pDecoder, void* pObj) {
   return code;
 }
 
+enum { PHY_STREAM_PARTITION_CODE_BASE_NODE = 1, PHY_STREAM_PARTITION_CODE_TAGS, PHY_STREAM_PARTITION_CODE_SUBTABLE };
+
+static int32_t physiStreamPartitionNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
+  const SStreamPartitionPhysiNode* pNode = (const SStreamPartitionPhysiNode*)pObj;
+
+  int32_t code = tlvEncodeObj(pEncoder, PHY_STREAM_PARTITION_CODE_BASE_NODE, physiPartitionNodeToMsg, &pNode->part);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, PHY_STREAM_PARTITION_CODE_TAGS, nodeListToMsg, pNode->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeObj(pEncoder, PHY_STREAM_PARTITION_CODE_SUBTABLE, nodeToMsg, pNode->pSubtable);
+  }
+
+  return code;
+}
+
+static int32_t msgToPhysiStreamPartitionNode(STlvDecoder* pDecoder, void* pObj) {
+  SStreamPartitionPhysiNode* pNode = (SStreamPartitionPhysiNode*)pObj;
+
+  int32_t code = TSDB_CODE_SUCCESS;
+  STlv*   pTlv = NULL;
+  tlvForEach(pDecoder, pTlv, code) {
+    switch (pTlv->type) {
+      case PHY_STREAM_PARTITION_CODE_BASE_NODE:
+        code = tlvDecodeObjFromTlv(pTlv, msgToPhysiPartitionNode, &pNode->part);
+        break;
+      case PHY_STREAM_PARTITION_CODE_TAGS:
+        code = msgToNodeListFromTlv(pTlv, (void**)&pNode->pTags);
+        break;
+      case PHY_STREAM_PARTITION_CODE_SUBTABLE:
+        code = msgToNodeFromTlv(pTlv, (void**)&pNode->pSubtable);
+        break;
+      default:
+        break;
+    }
+  }
+
+  return code;
+}
+
 enum { PHY_INDEF_ROWS_FUNC_CODE_BASE_NODE = 1, PHY_INDEF_ROWS_FUNC_CODE_EXPRS, PHY_INDEF_ROWS_FUNC_CODE_FUNCS };
 
 static int32_t physiIndefRowsFuncNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
@@ -3592,9 +3646,11 @@ static int32_t specificNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
       code = physiStateWindowNodeToMsg(pObj, pEncoder);
       break;
     case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
-    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
       code = physiPartitionNodeToMsg(pObj, pEncoder);
       break;
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
+      code = physiStreamPartitionNodeToMsg(pObj, pEncoder);
+      break;
     case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC:
       code = physiIndefRowsFuncNodeToMsg(pObj, pEncoder);
       break;
@@ -3727,9 +3783,11 @@ static int32_t msgToSpecificNode(STlvDecoder* pDecoder, void* pObj) {
       code = msgToPhysiStateWindowNode(pDecoder, pObj);
       break;
     case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
-    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
       code = msgToPhysiPartitionNode(pDecoder, pObj);
       break;
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
+      code = msgToPhysiStreamPartitionNode(pDecoder, pObj);
+      break;
     case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC:
       code = msgToPhysiIndefRowsFuncNode(pDecoder, pObj);
       break;

source/libs/nodes/src/nodesTraverseFuncs.c

@@ -378,6 +378,8 @@ void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker wa
       nodesWalkExpr(pSelect->pWhere, walker, pContext);
     case SQL_CLAUSE_WHERE:
       nodesWalkExprs(pSelect->pPartitionByList, walker, pContext);
+      nodesWalkExprs(pSelect->pTags, walker, pContext);
+      nodesWalkExpr(pSelect->pSubtable, walker, pContext);
     case SQL_CLAUSE_PARTITION_BY:
       nodesWalkExpr(pSelect->pWindow, walker, pContext);
     case SQL_CLAUSE_WINDOW:
@@ -412,6 +414,8 @@ void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewrit
       nodesRewriteExpr(&(pSelect->pWhere), rewriter, pContext);
     case SQL_CLAUSE_WHERE:
       nodesRewriteExprs(pSelect->pPartitionByList, rewriter, pContext);
+      nodesRewriteExprs(pSelect->pTags, rewriter, pContext);
+      nodesRewriteExpr(&(pSelect->pSubtable), rewriter, pContext);
     case SQL_CLAUSE_PARTITION_BY:
       nodesRewriteExpr(&(pSelect->pWindow), rewriter, pContext);
     case SQL_CLAUSE_WINDOW:

source/libs/nodes/src/nodesUtilFuncs.c

@@ -772,6 +772,8 @@ void nodesDestroyNode(SNode* pNode) {
       nodesDestroyNode(pStmt->pFromTable);
       nodesDestroyNode(pStmt->pWhere);
       nodesDestroyList(pStmt->pPartitionByList);
+      nodesDestroyList(pStmt->pTags);
+      nodesDestroyNode(pStmt->pSubtable);
       nodesDestroyNode(pStmt->pWindow);
       nodesDestroyList(pStmt->pGroupByList);
       nodesDestroyNode(pStmt->pHaving);

source/libs/parser/inc/parAst.h

@@ -147,7 +147,7 @@ SNode* createCreateDatabaseStmt(SAstCreateContext* pCxt, bool ignoreExists, STok
 SNode* createDropDatabaseStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SToken* pDbName);
 SNode* createAlterDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName, SNode* pOptions);
 SNode* createFlushDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName);
-SNode* createTrimDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName);
+SNode* createTrimDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName, int32_t maxSpeed);
 SNode* createDefaultTableOptions(SAstCreateContext* pCxt);
 SNode* createAlterTableOptions(SAstCreateContext* pCxt);
 SNode* setTableOption(SAstCreateContext* pCxt, SNode* pOptions, ETableOptionType type, void* pVal);
@@ -212,7 +212,7 @@ SNode* createCreateFunctionStmt(SAstCreateContext* pCxt, bool ignoreExists, bool
 SNode* createDropFunctionStmt(SAstCreateContext* pCxt, bool ignoreNotExists, const SToken* pFuncName);
 SNode* createStreamOptions(SAstCreateContext* pCxt);
 SNode* createCreateStreamStmt(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pStreamName, SNode* pRealTable,
-                              SNode* pOptions, SNode* pQuery);
+                              SNode* pOptions, SNodeList* pTags, SNode* pSubtable, SNode* pQuery);
 SNode* createDropStreamStmt(SAstCreateContext* pCxt, bool ignoreNotExists, const SToken* pStreamName);
 SNode* createKillStmt(SAstCreateContext* pCxt, ENodeType type, const SToken* pId);
 SNode* createKillQueryStmt(SAstCreateContext* pCxt, const SToken* pQueryId);

source/libs/parser/inc/sql.y

@@ -159,7 +159,7 @@ cmd ::= DROP DATABASE exists_opt(A) db_name(B).
 cmd ::= USE db_name(A).                                                           { pCxt->pRootNode = createUseDatabaseStmt(pCxt, &A); }
 cmd ::= ALTER DATABASE db_name(A) alter_db_options(B).                            { pCxt->pRootNode = createAlterDatabaseStmt(pCxt, &A, B); }
 cmd ::= FLUSH DATABASE db_name(A).                                                { pCxt->pRootNode = createFlushDatabaseStmt(pCxt, &A); }
-cmd ::= TRIM DATABASE db_name(A).                                                 { pCxt->pRootNode = createTrimDatabaseStmt(pCxt, &A); }
+cmd ::= TRIM DATABASE db_name(A) speed_opt(B).                                    { pCxt->pRootNode = createTrimDatabaseStmt(pCxt, &A, B); }
 
 %type not_exists_opt                                                              { bool }
 %destructor not_exists_opt                                                        { }
@@ -246,6 +246,11 @@ retention_list(A) ::= retention_list(B) NK_COMMA retention(C).
 
 retention(A) ::= NK_VARIABLE(B) NK_COLON NK_VARIABLE(C).                          { A = createNodeListNodeEx(pCxt, createDurationValueNode(pCxt, &B), createDurationValueNode(pCxt, &C)); }
 
+%type speed_opt                                                                   { int32_t }
+%destructor speed_opt                                                             { }
+speed_opt(A) ::= .                                                                { A = 0; }
+speed_opt(A) ::= MAX_SPEED NK_INTEGER(B).                                         { A = taosStr2Int32(B.z, NULL, 10); }
+
 /************************************************ create/drop table/stable ********************************************/
 cmd ::= CREATE TABLE not_exists_opt(A) full_table_name(B)
   NK_LP column_def_list(C) NK_RP tags_def_opt(D) table_options(E).                { pCxt->pRootNode = createCreateTableStmt(pCxt, A, B, C, D, E); }
@@ -501,8 +506,8 @@ bufsize_opt(A) ::= .
 bufsize_opt(A) ::= BUFSIZE NK_INTEGER(B).                                         { A = taosStr2Int32(B.z, NULL, 10); }
 
 /************************************************ create/drop stream **************************************************/
-cmd ::= CREATE STREAM not_exists_opt(E) stream_name(A)
+cmd ::= CREATE STREAM not_exists_opt(E) stream_name(A) stream_options(B) INTO
-  stream_options(B) INTO full_table_name(C) AS query_or_subquery(D).              { pCxt->pRootNode = createCreateStreamStmt(pCxt, E, &A, C, B, D); }
+  full_table_name(C) tags_def_opt(F) subtable_opt(G) AS query_or_subquery(D).     { pCxt->pRootNode = createCreateStreamStmt(pCxt, E, &A, C, B, F, G, D); }
 cmd ::= DROP STREAM exists_opt(A) stream_name(B).                                 { pCxt->pRootNode = createDropStreamStmt(pCxt, A, &B); }
 
 stream_options(A) ::= .                                                           { A = createStreamOptions(pCxt); }
@@ -512,6 +517,9 @@ stream_options(A) ::= stream_options(B) TRIGGER MAX_DELAY duration_literal(C).
 stream_options(A) ::= stream_options(B) WATERMARK duration_literal(C).            { ((SStreamOptions*)B)->pWatermark = releaseRawExprNode(pCxt, C); A = B; }
 stream_options(A) ::= stream_options(B) IGNORE EXPIRED NK_INTEGER(C).             { ((SStreamOptions*)B)->ignoreExpired = taosStr2Int8(C.z, NULL, 10); A = B; }
 
+subtable_opt(A) ::= .                                                             { A = NULL; }
+subtable_opt(A) ::= SUBTABLE NK_LP expression(B) NK_RP.                           { A = releaseRawExprNode(pCxt, B); }
+
 /************************************************ kill connection/query ***********************************************/
 cmd ::= KILL CONNECTION NK_INTEGER(A).                                            { pCxt->pRootNode = createKillStmt(pCxt, QUERY_NODE_KILL_CONNECTION_STMT, &A); }
 cmd ::= KILL QUERY NK_STRING(A).                                                  { pCxt->pRootNode = createKillQueryStmt(pCxt, &A); }
@@ -909,7 +917,16 @@ where_clause_opt(A) ::= WHERE search_condition(B).
 %type partition_by_clause_opt                                                     { SNodeList* }
 %destructor partition_by_clause_opt                                               { nodesDestroyList($$); }
 partition_by_clause_opt(A) ::= .                                                  { A = NULL; }
-partition_by_clause_opt(A) ::= PARTITION BY expression_list(B).                   { A = B; }
+partition_by_clause_opt(A) ::= PARTITION BY partition_list(B).                    { A = B; }
+
+%type partition_list                                                              { SNodeList* }
+%destructor partition_list                                                        { nodesDestroyList($$); }
+partition_list(A) ::= partition_item(B).                                          { A = createNodeList(pCxt, B); }
+partition_list(A) ::= partition_list(B) NK_COMMA partition_item(C).               { A = addNodeToList(pCxt, B, C); }
+
+partition_item(A) ::= expr_or_subquery(B).                                        { A = releaseRawExprNode(pCxt, B); }
+partition_item(A) ::= expr_or_subquery(B) column_alias(C).                        { A = setProjectionAlias(pCxt, releaseRawExprNode(pCxt, B), &C); }
+partition_item(A) ::= expr_or_subquery(B) AS column_alias(C).                     { A = setProjectionAlias(pCxt, releaseRawExprNode(pCxt, B), &C); }
 
 twindow_clause_opt(A) ::= .                                                       { A = NULL; }
 twindow_clause_opt(A) ::=

source/libs/parser/src/parAstCreater.c

@@ -1055,7 +1055,7 @@ SNode* createFlushDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName) {
   return (SNode*)pStmt;
 }
 
-SNode* createTrimDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName) {
+SNode* createTrimDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName, int32_t maxSpeed) {
   CHECK_PARSER_STATUS(pCxt);
   if (!checkDbName(pCxt, pDbName, false)) {
     return NULL;
@@ -1063,6 +1063,7 @@ SNode* createTrimDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName) {
   STrimDatabaseStmt* pStmt = (STrimDatabaseStmt*)nodesMakeNode(QUERY_NODE_TRIM_DATABASE_STMT);
   CHECK_OUT_OF_MEM(pStmt);
   COPY_STRING_FORM_ID_TOKEN(pStmt->dbName, pDbName);
+  pStmt->maxSpeed = maxSpeed;
   return (SNode*)pStmt;
 }
 
@@ -1700,7 +1701,7 @@ SNode* createStreamOptions(SAstCreateContext* pCxt) {
 }
 
 SNode* createCreateStreamStmt(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pStreamName, SNode* pRealTable,
-                              SNode* pOptions, SNode* pQuery) {
+                              SNode* pOptions, SNodeList* pTags, SNode* pSubtable, SNode* pQuery) {
   CHECK_PARSER_STATUS(pCxt);
   SCreateStreamStmt* pStmt = (SCreateStreamStmt*)nodesMakeNode(QUERY_NODE_CREATE_STREAM_STMT);
   CHECK_OUT_OF_MEM(pStmt);
@@ -1713,6 +1714,8 @@ SNode* createCreateStreamStmt(SAstCreateContext* pCxt, bool ignoreExists, const
   pStmt->ignoreExists = ignoreExists;
   pStmt->pOptions = (SStreamOptions*)pOptions;
   pStmt->pQuery = pQuery;
+  pStmt->pTags = pTags;
+  pStmt->pSubtable = pSubtable;
   return (SNode*)pStmt;
 }
 

source/libs/parser/src/parCalcConst.c

@@ -275,6 +275,12 @@ static int32_t calcConstSelectFrom(SCalcConstContext* pCxt, SSelectStmt* pSelect
   if (TSDB_CODE_SUCCESS == code) {
     code = calcConstList(pSelect->pPartitionByList);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = calcConstList(pSelect->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = calcConstNode(&pSelect->pSubtable);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = calcConstNode(&pSelect->pWindow);
   }

source/libs/parser/src/parTokenizer.c

@@ -129,6 +129,7 @@ static SKeyword keywordTable[] = {
     {"MATCH",                TK_MATCH},
     {"MAXROWS",              TK_MAXROWS},
     {"MAX_DELAY",            TK_MAX_DELAY},
+    {"MAX_SPEED",            TK_MAX_SPEED},
     {"MERGE",                TK_MERGE},
     {"META",                 TK_META},
     {"MINROWS",              TK_MINROWS},
@@ -200,6 +201,7 @@ static SKeyword keywordTable[] = {
     {"STREAMS",              TK_STREAMS},
     {"STRICT",               TK_STRICT},
     {"SUBSCRIPTIONS",        TK_SUBSCRIPTIONS},
+    {"SUBTABLE",             TK_SUBTABLE},
     {"SYSINFO",              TK_SYSINFO},
     {"TABLE",                TK_TABLE},
     {"TABLES",               TK_TABLES},

source/libs/parser/src/parTranslater.c

@@ -3076,7 +3076,14 @@ static int32_t translatePartitionBy(STranslateContext* pCxt, SSelectStmt* pSelec
     return TSDB_CODE_SUCCESS;
   }
   pCxt->currClause = SQL_CLAUSE_PARTITION_BY;
-  return translateExprList(pCxt, pSelect->pPartitionByList);
+  int32_t code = translateExprList(pCxt, pSelect->pPartitionByList);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = translateExprList(pCxt, pSelect->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = translateExpr(pCxt, &pSelect->pSubtable);
+  }
+  return code;
 }
 
 static int32_t translateWhere(STranslateContext* pCxt, SSelectStmt* pSelect) {
@@ -3968,7 +3975,7 @@ static int32_t translateAlterDatabase(STranslateContext* pCxt, SAlterDatabaseStm
 }
 
 static int32_t translateTrimDatabase(STranslateContext* pCxt, STrimDatabaseStmt* pStmt) {
-  STrimDbReq req = {0};
+  STrimDbReq req = {.maxSpeed = pStmt->maxSpeed};
   SName      name = {0};
   tNameSetDbName(&name, pCxt->pParseCxt->acctId, pStmt->dbName, strlen(pStmt->dbName));
   tNameGetFullDbName(&name, req.db);
@@ -5211,6 +5218,93 @@ static int32_t addWstartTsToCreateStreamQuery(SNode* pStmt) {
   return code;
 }
 
+static int32_t addTagsToCreateStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStmt, SSelectStmt* pSelect) {
+  if (NULL == pStmt->pTags) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SNode* pTag = NULL;
+  FOREACH(pTag, pStmt->pTags) {
+    bool   found = false;
+    SNode* pPart = NULL;
+    FOREACH(pPart, pSelect->pPartitionByList) {
+      if (0 == strcmp(((SColumnDefNode*)pTag)->colName, ((SExprNode*)pPart)->userAlias)) {
+        if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSelect->pTags, nodesCloneNode(pPart))) {
+          return TSDB_CODE_OUT_OF_MEMORY;
+        }
+        found = true;
+        break;
+      }
+    }
+    if (!found) {
+      return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_COLUMN, ((SColumnDefNode*)pTag)->colName);
+    }
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+typedef struct SRewriteSubtableCxt {
+  STranslateContext* pCxt;
+  SNodeList*         pPartitionList;
+} SRewriteSubtableCxt;
+
+static EDealRes rewriteSubtable(SNode** pNode, void* pContext) {
+  if (QUERY_NODE_COLUMN == nodeType(*pNode)) {
+    SRewriteSubtableCxt* pCxt = pContext;
+    bool                 found = false;
+    SNode*               pPart = NULL;
+    FOREACH(pPart, pCxt->pPartitionList) {
+      if (0 == strcmp(((SColumnNode*)*pNode)->colName, ((SExprNode*)pPart)->userAlias)) {
+        SNode* pNew = nodesCloneNode(pPart);
+        if (NULL == pNew) {
+          pCxt->pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
+          return DEAL_RES_ERROR;
+        }
+        nodesDestroyNode(*pNode);
+        *pNode = pNew;
+        found = true;
+        break;
+      }
+      if (!found) {
+        return generateDealNodeErrMsg(pCxt->pCxt, TSDB_CODE_PAR_INVALID_COLUMN, ((SColumnNode*)*pNode)->colName);
+      }
+    }
+    return DEAL_RES_IGNORE_CHILD;
+  }
+  return DEAL_RES_CONTINUE;
+}
+
+static int32_t addSubtableNameToCreateStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStmt,
+                                                  SSelectStmt* pSelect) {
+  if (NULL == pStmt->pSubtable) {
+    return TSDB_CODE_SUCCESS;
+  }
+  pSelect->pSubtable = nodesCloneNode(pStmt->pSubtable);
+  if (NULL == pSelect->pSubtable) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  SRewriteSubtableCxt cxt = {.pCxt = pCxt, .pPartitionList = pSelect->pPartitionByList};
+  nodesRewriteExpr(&pSelect->pSubtable, rewriteSubtable, &cxt);
+  return pCxt->errCode;
+}
+
+static int32_t addSubtableInfoToCreateStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStmt) {
+  SSelectStmt* pSelect = (SSelectStmt*)pStmt->pQuery;
+  if (NULL == pSelect->pPartitionByList) {
+    if (NULL != pStmt->pTags || NULL != pStmt->pSubtable) {
+      return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Unsupported stream query");
+    }
+    return TSDB_CODE_SUCCESS;
+  }
+
+  int32_t code = addTagsToCreateStreamQuery(pCxt, pStmt, pSelect);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = addSubtableNameToCreateStreamQuery(pCxt, pStmt, pSelect);
+  }
+
+  return code;
+}
+
 static int32_t checkStreamQuery(STranslateContext* pCxt, SSelectStmt* pSelect) {
   if (TSDB_DATA_TYPE_TIMESTAMP != ((SExprNode*)nodesListGetNode(pSelect->pProjectionList, 0))->resType.type ||
       !pSelect->isTimeLineResult || crossTableWithoutAggOper(pSelect) || NULL != pSelect->pOrderByList ||
@@ -5220,18 +5314,21 @@ static int32_t checkStreamQuery(STranslateContext* pCxt, SSelectStmt* pSelect) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t buildCreateStreamQuery(STranslateContext* pCxt, SNode* pStmt, SCMCreateStreamReq* pReq) {
+static int32_t buildCreateStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStmt, SCMCreateStreamReq* pReq) {
   pCxt->createStream = true;
-  int32_t code = addWstartTsToCreateStreamQuery(pStmt);
+  int32_t code = addWstartTsToCreateStreamQuery(pStmt->pQuery);
   if (TSDB_CODE_SUCCESS == code) {
-    code = translateQuery(pCxt, pStmt);
+    code = addSubtableInfoToCreateStreamQuery(pCxt, pStmt);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = checkStreamQuery(pCxt, (SSelectStmt*)pStmt);
+    code = translateQuery(pCxt, pStmt->pQuery);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    getSourceDatabase(pStmt, pCxt->pParseCxt->acctId, pReq->sourceDB);
+    code = checkStreamQuery(pCxt, (SSelectStmt*)pStmt->pQuery);
-    code = nodesNodeToString(pStmt, false, &pReq->ast, NULL);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    getSourceDatabase(pStmt->pQuery, pCxt->pParseCxt->acctId, pReq->sourceDB);
+    code = nodesNodeToString(pStmt->pQuery, false, &pReq->ast, NULL);
   }
   return code;
 }
@@ -5249,7 +5346,7 @@ static int32_t buildCreateStreamReq(STranslateContext* pCxt, SCreateStreamStmt*
     tNameExtractFullName(&name, pReq->targetStbFullName);
   }
 
-  int32_t code = buildCreateStreamQuery(pCxt, pStmt->pQuery, pReq);
+  int32_t code = buildCreateStreamQuery(pCxt, pStmt, pReq);
   if (TSDB_CODE_SUCCESS == code) {
     pReq->sql = strdup(pCxt->pParseCxt->pSql);
     if (NULL == pReq->sql) {
@@ -5262,6 +5359,8 @@ static int32_t buildCreateStreamReq(STranslateContext* pCxt, SCreateStreamStmt*
     pReq->maxDelay = (NULL != pStmt->pOptions->pDelay ? ((SValueNode*)pStmt->pOptions->pDelay)->datum.i : 0);
     pReq->watermark = (NULL != pStmt->pOptions->pWatermark ? ((SValueNode*)pStmt->pOptions->pWatermark)->datum.i : 0);
     pReq->igExpired = pStmt->pOptions->ignoreExpired;
+    columnDefNodeToField(pStmt->pTags, &pReq->pTags);
+    pReq->numOfTags = LIST_LENGTH(pStmt->pTags);
   }
 
   return code;

source/libs/parser/test/parInitialCTest.cpp

@@ -610,6 +610,20 @@ TEST_F(ParserInitialCTest, createStream) {
     expect.igExpired = igExpired;
   };
 
+  auto addTag = [&](const char* pFieldName, uint8_t type, int32_t bytes = 0) {
+    SField field = {0};
+    strcpy(field.name, pFieldName);
+    field.type = type;
+    field.bytes = bytes > 0 ? bytes : tDataTypes[type].bytes;
+    field.flags |= COL_SMA_ON;
+
+    if (NULL == expect.pTags) {
+      expect.pTags = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SField));
+    }
+    taosArrayPush(expect.pTags, &field);
+    expect.numOfTags += 1;
+  };
+
   setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
     ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_CREATE_STREAM_STMT);
     SCMCreateStreamReq req = {0};
@@ -625,6 +639,19 @@ TEST_F(ParserInitialCTest, createStream) {
     ASSERT_EQ(req.maxDelay, expect.maxDelay);
     ASSERT_EQ(req.watermark, expect.watermark);
     ASSERT_EQ(req.igExpired, expect.igExpired);
+    ASSERT_EQ(req.numOfTags, expect.numOfTags);
+    if (expect.numOfTags > 0) {
+      ASSERT_EQ(taosArrayGetSize(req.pTags), expect.numOfTags);
+      ASSERT_EQ(taosArrayGetSize(req.pTags), taosArrayGetSize(expect.pTags));
+      for (int32_t i = 0; i < expect.numOfTags; ++i) {
+        SField* pField = (SField*)taosArrayGet(req.pTags, i);
+        SField* pExpectField = (SField*)taosArrayGet(expect.pTags, i);
+        ASSERT_EQ(std::string(pField->name), std::string(pExpectField->name));
+        ASSERT_EQ(pField->type, pExpectField->type);
+        ASSERT_EQ(pField->bytes, pExpectField->bytes);
+        ASSERT_EQ(pField->flags, pExpectField->flags);
+      }
+    }
     tFreeSCMCreateStreamReq(&req);
   });
 
@@ -640,6 +667,17 @@ TEST_F(ParserInitialCTest, createStream) {
   run("CREATE STREAM IF NOT EXISTS s1 TRIGGER MAX_DELAY 20s WATERMARK 10s IGNORE EXPIRED 0 INTO st1 AS SELECT COUNT(*) "
       "FROM t1 INTERVAL(10S)");
   clearCreateStreamReq();
+
+  setCreateStreamReqFunc(
+      "s1", "test",
+      "create stream s1 into st3 tags(tname varchar(10), id int) subtable(concat('new-', tname)) as "
+      "select _wstart wstart, count(*) cnt from st1 partition by tbname tname, tag1 id interval(10s)",
+      "st3");
+  addTag("tname", TSDB_DATA_TYPE_VARCHAR, 10 + VARSTR_HEADER_SIZE);
+  addTag("id", TSDB_DATA_TYPE_INT);
+  run("CREATE STREAM s1 INTO st3 TAGS(tname VARCHAR(10), id INT) SUBTABLE(CONCAT('new-', tname)) "
+      "AS SELECT _WSTART wstart, COUNT(*) cnt FROM st1 PARTITION BY TBNAME tname, tag1 id INTERVAL(10S)");
+  clearCreateStreamReq();
 }
 
 TEST_F(ParserInitialCTest, createStreamSemanticCheck) {

source/libs/parser/test/parShowToUse.cpp

@@ -250,7 +250,10 @@ TEST_F(ParserShowToUseTest, trimDatabase) {
 
   STrimDbReq expect = {0};
 
-  auto setTrimDbReq = [&](const char* pDb) { snprintf(expect.db, sizeof(expect.db), "0.%s", pDb); };
+  auto setTrimDbReq = [&](const char* pDb, int32_t maxSpeed = 0) {
+    snprintf(expect.db, sizeof(expect.db), "0.%s", pDb);
+    expect.maxSpeed = maxSpeed;
+  };
 
   setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
     ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_TRIM_DATABASE_STMT);
@@ -258,10 +261,14 @@ TEST_F(ParserShowToUseTest, trimDatabase) {
     STrimDbReq req = {0};
     ASSERT_EQ(tDeserializeSTrimDbReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req), TSDB_CODE_SUCCESS);
     ASSERT_EQ(std::string(req.db), std::string(expect.db));
+    ASSERT_EQ(req.maxSpeed, expect.maxSpeed);
   });
 
   setTrimDbReq("wxy_db");
   run("TRIM DATABASE wxy_db");
+
+  setTrimDbReq("wxy_db", 100);
+  run("TRIM DATABASE wxy_db MAX_SPEED 100");
 }
 
 TEST_F(ParserShowToUseTest, useDatabase) {

source/libs/planner/src/planLogicCreater.c

@@ -1022,6 +1022,20 @@ static int32_t createPartitionLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pS
     }
   }
 
+  if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pTags) {
+    pPartition->pTags = nodesCloneList(pSelect->pTags);
+    if (NULL == pPartition->pTags) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
+  if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pSubtable) {
+    pPartition->pSubtable = nodesCloneNode(pSelect->pSubtable);
+    if (NULL == pPartition->pSubtable) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
   if (TSDB_CODE_SUCCESS == code) {
     *pLogicNode = (SLogicNode*)pPartition;
   } else {

source/libs/planner/src/planOptimizer.c

@@ -1615,6 +1615,8 @@ static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSub
   SScanLogicNode* pScan = (SScanLogicNode*)nodesListGetNode(pNode->pChildren, 0);
   if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
     TSWAP(((SPartitionLogicNode*)pNode)->pPartitionKeys, pScan->pGroupTags);
+    TSWAP(((SPartitionLogicNode*)pNode)->pTags, pScan->pTags);
+    TSWAP(((SPartitionLogicNode*)pNode)->pSubtable, pScan->pSubtable);
     int32_t code = replaceLogicNode(pLogicSubplan, pNode, (SLogicNode*)pScan);
     if (TSDB_CODE_SUCCESS == code) {
       code = adjustLogicNodeDataRequirement((SLogicNode*)pScan, pNode->resultDataOrder);

source/libs/planner/src/planPhysiCreater.c

@@ -563,7 +563,16 @@ static int32_t createTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubp
   pTableScan->igExpired = pScanLogicNode->igExpired;
   pTableScan->assignBlockUid = pCxt->pPlanCxt->rSmaQuery ? true : false;
 
-  return createScanPhysiNodeFinalize(pCxt, pSubplan, pScanLogicNode, (SScanPhysiNode*)pTableScan, pPhyNode);
+  int32_t code = createScanPhysiNodeFinalize(pCxt, pSubplan, pScanLogicNode, (SScanPhysiNode*)pTableScan, pPhyNode);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setListSlotId(pCxt, pTableScan->scan.node.pOutputDataBlockDesc->dataBlockId, -1, pScanLogicNode->pTags,
+                         &pTableScan->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setNodeSlotId(pCxt, pTableScan->scan.node.pOutputDataBlockDesc->dataBlockId, -1, pScanLogicNode->pSubtable,
+                         &pTableScan->pSubtable);
+  }
+  return code;
 }
 
 static int32_t createSystemTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan,
@@ -1322,11 +1331,10 @@ static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
   return code;
 }
 
-static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren,
+static int32_t createPartitionPhysiNodeImpl(SPhysiPlanContext* pCxt, SNodeList* pChildren,
-                                        SPartitionLogicNode* pPartLogicNode, SPhysiNode** pPhyNode) {
+                                            SPartitionLogicNode* pPartLogicNode, ENodeType type,
-  SPartitionPhysiNode* pPart = (SPartitionPhysiNode*)makePhysiNode(
+                                            SPhysiNode** pPhyNode) {
-      pCxt, (SLogicNode*)pPartLogicNode,
+  SPartitionPhysiNode* pPart = (SPartitionPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pPartLogicNode, type);
-      pCxt->pPlanCxt->streamQuery ? QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION : QUERY_NODE_PHYSICAL_PLAN_PARTITION);
   if (NULL == pPart) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
@@ -1371,6 +1379,34 @@ static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChi
   return code;
 }
 
+static int32_t createStreamPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren,
+                                              SPartitionLogicNode* pPartLogicNode, SPhysiNode** pPhyNode) {
+  SStreamPartitionPhysiNode* pPart = NULL;
+  int32_t                    code = createPartitionPhysiNodeImpl(pCxt, pChildren, pPartLogicNode,
+                                                                 QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION, (SPhysiNode**)&pPart);
+  SDataBlockDescNode*        pChildTupe = (((SPhysiNode*)nodesListGetNode(pChildren, 0))->pOutputDataBlockDesc);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pPartLogicNode->pTags, &pPart->pTags);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setNodeSlotId(pCxt, pChildTupe->dataBlockId, -1, pPartLogicNode->pSubtable, &pPart->pSubtable);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    *pPhyNode = (SPhysiNode*)pPart;
+  } else {
+    nodesDestroyNode((SNode*)pPart);
+  }
+  return code;
+}
+
+static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren,
+                                        SPartitionLogicNode* pPartLogicNode, SPhysiNode** pPhyNode) {
+  if (pCxt->pPlanCxt->streamQuery) {
+    return createStreamPartitionPhysiNode(pCxt, pChildren, pPartLogicNode, pPhyNode);
+  }
+  return createPartitionPhysiNodeImpl(pCxt, pChildren, pPartLogicNode, QUERY_NODE_PHYSICAL_PLAN_PARTITION, pPhyNode);
+}
+
 static int32_t createFillPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SFillLogicNode* pFillNode,
                                    SPhysiNode** pPhyNode) {
   SFillPhysiNode* pFill = (SFillPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pFillNode, QUERY_NODE_PHYSICAL_PLAN_FILL);

source/libs/planner/test/planOtherTest.cpp

@@ -32,6 +32,12 @@ TEST_F(PlanOtherTest, createStream) {
 
   run("create stream if not exists s1 trigger window_close watermark 10s into st1 as select count(*) from t1 "
       "interval(10s)");
+
+  run("CREATE STREAM s1 INTO st3 TAGS(tname VARCHAR(10), id INT) SUBTABLE(CONCAT('new-', tname)) "
+      "AS SELECT _WSTART wstart, COUNT(*) cnt FROM st1 PARTITION BY TBNAME tname, c1 id INTERVAL(10S)");
+
+  run("CREATE STREAM s1 INTO st3 TAGS(tname VARCHAR(10), id INT) SUBTABLE(CONCAT('new-', tname)) "
+      "AS SELECT _WSTART wstart, COUNT(*) cnt FROM st1 PARTITION BY TBNAME tname, tag1 id INTERVAL(10S)");
 }
 
 TEST_F(PlanOtherTest, createStreamUseSTable) {

source/libs/sync/src/syncUtil.c

@@ -57,7 +57,7 @@ void syncUtilU642Addr(uint64_t u64, char* host, size_t len, uint16_t* port) {
 
   struct in_addr addr;
   addr.s_addr = hostU32;
-  snprintf(host, len, "%s", taosInetNtoa(addr));
+  taosInetNtoa(addr, host, len);
   *port = (uint16_t)((u64 & 0x00000000FFFF0000) >> 16);
 }
 

source/os/src/osDir.c

@@ -58,6 +58,24 @@ int wordexp(char *words, wordexp_t *pwordexp, int flags) {
 
 void wordfree(wordexp_t *pwordexp) {}
 
+#elif defined(DARWIN)
+
+#include <dirent.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#include <wordexp.h>
+
+typedef struct dirent dirent;
+typedef struct dirent TdDirEntry;
+
+typedef struct TdDir {
+  TdDirEntry    dirEntry;
+  TdDirEntry    dirEntry1;
+  TdDirEntryPtr dirEntryPtr;
+  DIR          *pDir;
+} TdDir;
+
 #else
 
 #include <dirent.h>
@@ -343,7 +361,7 @@ char *taosDirName(char *name) {
     name[0] = 0;
   }
   return name;
-#elif defined(_TD_DARWIN_64)
+#else
   char *end = strrchr(name, '/');
   if (end != NULL) {
     *end = '\0';
@@ -351,8 +369,6 @@ char *taosDirName(char *name) {
     name[0] = 0;
   }
   return name;
-#else
-  return dirname(name);
 #endif
 }
 
@@ -362,7 +378,9 @@ char *taosDirEntryBaseName(char *name) {
   _splitpath(name, NULL, NULL, Filename1, Ext1);
   return name + (strlen(name) - strlen(Filename1) - strlen(Ext1));
 #else
-  return (char *)basename(name);
+  char *pPoint = strchr(name, '.');
+  if (pPoint != NULL) pPoint = 0;
+  return name;
 #endif
 }
 
@@ -386,6 +404,13 @@ TdDirPtr taosOpenDir(const char *dirname) {
     return NULL;
   }
   return pDir;
+#elif defined(DARWIN)
+  DIR *pDir = opendir(dirname);
+  if (pDir == NULL) return NULL;
+  TdDirPtr dirPtr = (TdDirPtr)taosMemoryMalloc(sizeof(TdDir));
+  dirPtr->dirEntryPtr = (TdDirEntryPtr)&(dirPtr->dirEntry1);
+  dirPtr->pDir = pDir;
+  return dirPtr;
 #else
   return (TdDirPtr)opendir(dirname);
 #endif
@@ -400,6 +425,12 @@ TdDirEntryPtr taosReadDir(TdDirPtr pDir) {
     return NULL;
   }
   return (TdDirEntryPtr) & (pDir->dirEntry.findFileData);
+#elif defined(DARWIN)
+  if (readdir_r(pDir->pDir, (dirent*)&(pDir->dirEntry), (dirent**)&(pDir->dirEntryPtr)) == 0) {
+    return pDir->dirEntryPtr;
+  } else {
+    return NULL;
+  }
 #else
   return (TdDirEntryPtr)readdir((DIR *)pDir);
 #endif
@@ -436,6 +467,11 @@ int32_t taosCloseDir(TdDirPtr *ppDir) {
   taosMemoryFree(*ppDir);
   *ppDir = NULL;
   return 0;
+#elif defined(DARWIN)
+  closedir((*ppDir)->pDir);
+  taosMemoryFree(*ppDir);
+  *ppDir = NULL;
+  return 0;
 #else
   closedir((DIR *)*ppDir);
   *ppDir = NULL;

source/os/src/osSemaphore.c

@@ -106,293 +106,8 @@ int32_t tsem_timewait(tsem_t* sem, int64_t nanosecs) {
 
 #elif defined(_TD_DARWIN_64)
 
-/*
- * darwin implementation
- */
-
 #include <libproc.h>
 
-// #define SEM_USE_PTHREAD
-// #define SEM_USE_POSIX
-// #define SEM_USE_SEM
-
-// #ifdef SEM_USE_SEM
-// #include <mach/mach_error.h>
-// #include <mach/mach_init.h>
-// #include <mach/semaphore.h>
-// #include <mach/task.h>
-
-// static TdThread     sem_thread;
-// static TdThreadOnce sem_once;
-// static task_t       sem_port;
-// static volatile int sem_inited = 0;
-// static semaphore_t  sem_exit;
-
-// static void *sem_thread_routine(void *arg) {
-//   (void)arg;
-//   setThreadName("sem_thrd");
-
-//   sem_port = mach_task_self();
-//   kern_return_t ret = semaphore_create(sem_port, &sem_exit, SYNC_POLICY_FIFO, 0);
-//   if (ret != KERN_SUCCESS) {
-//     fprintf(stderr, "==%s[%d]%s()==failed to create sem_exit\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__);
-//     sem_inited = -1;
-//     return NULL;
-//   }
-//   sem_inited = 1;
-//   semaphore_wait(sem_exit);
-//   return NULL;
-// }
-
-// static void once_init(void) {
-//   int r = 0;
-//   r = taosThreadCreate(&sem_thread, NULL, sem_thread_routine, NULL);
-//   if (r) {
-//     fprintf(stderr, "==%s[%d]%s()==failed to create thread\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__);
-//     return;
-//   }
-//   while (sem_inited == 0) {
-//     ;
-//   }
-// }
-// #endif
-
-// struct tsem_s {
-// #ifdef SEM_USE_PTHREAD
-//   TdThreadMutex    lock;
-//   TdThreadCond     cond;
-//   volatile int64_t val;
-// #elif defined(SEM_USE_POSIX)
-//   size_t        id;
-//   sem_t        *sem;
-// #elif defined(SEM_USE_SEM)
-//   semaphore_t sem;
-// #else   // SEM_USE_PTHREAD
-//   dispatch_semaphore_t sem;
-// #endif  // SEM_USE_PTHREAD
-
-//   volatile unsigned int valid : 1;
-// };
-
-// int tsem_init(tsem_t *sem, int pshared, unsigned int value) {
-//   // fprintf(stderr, "==%s[%d]%s():[%p]==creating\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
-//   if (*sem) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==already initialized\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   struct tsem_s *p = (struct tsem_s *)taosMemoryCalloc(1, sizeof(*p));
-//   if (!p) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==out of memory\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
-//     abort();
-//   }
-
-// #ifdef SEM_USE_PTHREAD
-//   int r = taosThreadMutexInit(&p->lock, NULL);
-//   do {
-//     if (r) break;
-//     r = taosThreadCondInit(&p->cond, NULL);
-//     if (r) {
-//       taosThreadMutexDestroy(&p->lock);
-//       break;
-//     }
-//     p->val = value;
-//   } while (0);
-//   if (r) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==not created\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
-//     abort();
-//   }
-// #elif defined(SEM_USE_POSIX)
-//   static size_t tick = 0;
-//   do {
-//     size_t id = atomic_add_fetch_64(&tick, 1);
-//     if (id == SEM_VALUE_MAX) {
-//       atomic_store_64(&tick, 0);
-//       id = 0;
-//     }
-//     char name[NAME_MAX - 4];
-//     snprintf(name, sizeof(name), "/t" PRId64, id);
-//     p->sem = sem_open(name, O_CREAT | O_EXCL, pshared, value);
-//     p->id = id;
-//     if (p->sem != SEM_FAILED) break;
-//     int e = errno;
-//     if (e == EEXIST) continue;
-//     if (e == EINTR) continue;
-//     fprintf(stderr, "==%s[%d]%s():[%p]==not created[%d]%s\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//     sem,
-//             e, strerror(e));
-//     abort();
-//   } while (p->sem == SEM_FAILED);
-// #elif defined(SEM_USE_SEM)
-//   taosThreadOnce(&sem_once, once_init);
-//   if (sem_inited != 1) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal resource init failed\n", taosDirEntryBaseName(__FILE__), __LINE__,
-//             __func__, sem);
-//     errno = ENOMEM;
-//     return -1;
-//   }
-//   kern_return_t ret = semaphore_create(sem_port, &p->sem, SYNC_POLICY_FIFO, value);
-//   if (ret != KERN_SUCCESS) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==semophore_create failed\n", taosDirEntryBaseName(__FILE__), __LINE__,
-//     __func__,
-//             sem);
-//     // we fail-fast here, because we have less-doc about semaphore_create for the moment
-//     abort();
-//   }
-// #else   // SEM_USE_PTHREAD
-//   p->sem = dispatch_semaphore_create(value);
-//   if (p->sem == NULL) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==not created\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
-//     abort();
-//   }
-// #endif  // SEM_USE_PTHREAD
-
-//   p->valid = 1;
-
-//   *sem = p;
-
-//   return 0;
-// }
-
-// int tsem_wait(tsem_t *sem) {
-//   if (!*sem) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
-//     abort();
-//   }
-//   struct tsem_s *p = *sem;
-//   if (!p->valid) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//     sem); abort();
-//   }
-// #ifdef SEM_USE_PTHREAD
-//   if (taosThreadMutexLock(&p->lock)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   p->val -= 1;
-//   if (p->val < 0) {
-//     if (taosThreadCondWait(&p->cond, &p->lock)) {
-//       fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__,
-//       __func__,
-//               sem);
-//       abort();
-//     }
-//   }
-//   if (taosThreadMutexUnlock(&p->lock)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   return 0;
-// #elif defined(SEM_USE_POSIX)
-//   return sem_wait(p->sem);
-// #elif defined(SEM_USE_SEM)
-//   return semaphore_wait(p->sem);
-// #else   // SEM_USE_PTHREAD
-//   return dispatch_semaphore_wait(p->sem, DISPATCH_TIME_FOREVER);
-// #endif  // SEM_USE_PTHREAD
-// }
-
-// int tsem_post(tsem_t *sem) {
-//   if (!*sem) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
-//     abort();
-//   }
-//   struct tsem_s *p = *sem;
-//   if (!p->valid) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//     sem); abort();
-//   }
-// #ifdef SEM_USE_PTHREAD
-//   if (taosThreadMutexLock(&p->lock)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   p->val += 1;
-//   if (p->val <= 0) {
-//     if (taosThreadCondSignal(&p->cond)) {
-//       fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__,
-//       __func__,
-//               sem);
-//       abort();
-//     }
-//   }
-//   if (taosThreadMutexUnlock(&p->lock)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   return 0;
-// #elif defined(SEM_USE_POSIX)
-//   return sem_post(p->sem);
-// #elif defined(SEM_USE_SEM)
-//   return semaphore_signal(p->sem);
-// #else   // SEM_USE_PTHREAD
-//   return dispatch_semaphore_signal(p->sem);
-// #endif  // SEM_USE_PTHREAD
-// }
-
-// int tsem_destroy(tsem_t *sem) {
-//   // fprintf(stderr, "==%s[%d]%s():[%p]==destroying\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__, sem);
-//   if (!*sem) {
-//     // fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//     sem);
-//     // abort();
-//     return 0;
-//   }
-//   struct tsem_s *p = *sem;
-//   if (!p->valid) {
-//     // fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//     // sem); abort();
-//     return 0;
-//   }
-// #ifdef SEM_USE_PTHREAD
-//   if (taosThreadMutexLock(&p->lock)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   p->valid = 0;
-//   if (taosThreadCondDestroy(&p->cond)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   if (taosThreadMutexUnlock(&p->lock)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-//   if (taosThreadMutexDestroy(&p->lock)) {
-//     fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//             sem);
-//     abort();
-//   }
-// #elif defined(SEM_USE_POSIX)
-//   char name[NAME_MAX - 4];
-//   snprintf(name, sizeof(name), "/t" PRId64, p->id);
-//   int r = sem_unlink(name);
-//   if (r) {
-//     int e = errno;
-//     fprintf(stderr, "==%s[%d]%s():[%p]==unlink failed[%d]%s\n", taosDirEntryBaseName(__FILE__), __LINE__, __func__,
-//     sem,
-//             e, strerror(e));
-//     abort();
-//   }
-// #elif defined(SEM_USE_SEM)
-//   semaphore_destroy(sem_port, p->sem);
-// #else   // SEM_USE_PTHREAD
-// #endif  // SEM_USE_PTHREAD
-
-//   p->valid = 0;
-//   taosMemoryFree(p);
-
-//   *sem = NULL;
-//   return 0;
-// }
-
 int tsem_init(tsem_t *psem, int flags, unsigned int count) {
   *psem = dispatch_semaphore_create(count);
   if (*psem == NULL) return -1;
@@ -401,8 +116,8 @@ int tsem_init(tsem_t *psem, int flags, unsigned int count) {
 
 int tsem_destroy(tsem_t *psem) {
   if (psem == NULL || *psem == NULL) return -1;
-  dispatch_release(*psem);
+  // dispatch_release(*psem);
-  *psem = NULL;
+  // *psem = NULL;
   return 0;
 }
 

source/os/src/osSocket.c

@@ -312,14 +312,8 @@ uint32_t taosInetAddr(const char *ipAddr) {
   return inet_addr(ipAddr);
 #endif
 }
-const char *taosInetNtoa(struct in_addr ipInt) {
+const char *taosInetNtoa(struct in_addr ipInt, char *dstStr, int32_t len) {
-#ifdef WINDOWS
+  return inet_ntop(AF_INET, &ipInt, dstStr, len);
-  // not thread safe, only for debug usage while print log
-  static char tmpDstStr[16];
-  return inet_ntop(AF_INET, &ipInt, tmpDstStr, INET6_ADDRSTRLEN);
-#else
-  return inet_ntoa(ipInt);
-#endif
 }
 
 #ifndef SIGPIPE
@@ -670,7 +664,7 @@ int taosGetLocalIp(const char *eth, char *ip) {
     return -1;
   }
   memcpy(&sin, &ifr.ifr_addr, sizeof(sin));
-  snprintf(ip, 64, "%s", inet_ntoa(sin.sin_addr));
+  taosInetNtoa(sin.sin_addr, ip, 64);
   taosCloseSocketNoCheck1(fd);
 #endif
   return 0;

source/os/src/osString.c

@@ -306,8 +306,6 @@ int32_t taosMbsToWchars(TdWchar *pWchars, const char *pStrs, int32_t size) { ret
 
 int32_t taosWcharToMb(char *pStr, TdWchar wchar) { return wctomb(pStr, wchar); }
 
-int32_t taosWcharsToMbs(char *pStrs, TdWchar *pWchars, int32_t size) { return wcstombs(pStrs, pWchars, size); }
-
 char *taosStrCaseStr(const char *str, const char *pattern) {
   size_t i;
 
@@ -326,6 +324,9 @@ char *taosStrCaseStr(const char *str, const char *pattern) {
 
 int64_t taosStr2Int64(const char *str, char **pEnd, int32_t radix) {
   int64_t tmp = strtoll(str, pEnd, radix);
+#ifdef DARWIN
+  if (errno == EINVAL) errno = 0;
+#endif
 #ifdef TD_CHECK_STR_TO_INT_ERROR
   assert(errno != ERANGE);
   assert(errno != EINVAL);
@@ -335,6 +336,9 @@ int64_t taosStr2Int64(const char *str, char **pEnd, int32_t radix) {
 
 uint64_t taosStr2UInt64(const char *str, char **pEnd, int32_t radix) {
   uint64_t tmp = strtoull(str, pEnd, radix);
+#ifdef DARWIN
+  if (errno == EINVAL) errno = 0;
+#endif
 #ifdef TD_CHECK_STR_TO_INT_ERROR
   assert(errno != ERANGE);
   assert(errno != EINVAL);
@@ -344,6 +348,9 @@ uint64_t taosStr2UInt64(const char *str, char **pEnd, int32_t radix) {
 
 int32_t taosStr2Int32(const char *str, char **pEnd, int32_t radix) {
   int32_t tmp = strtol(str, pEnd, radix);
+#ifdef DARWIN
+  if (errno == EINVAL) errno = 0;
+#endif
 #ifdef TD_CHECK_STR_TO_INT_ERROR
   assert(errno != ERANGE);
   assert(errno != EINVAL);
@@ -353,6 +360,9 @@ int32_t taosStr2Int32(const char *str, char **pEnd, int32_t radix) {
 
 uint32_t taosStr2UInt32(const char *str, char **pEnd, int32_t radix) {
   uint32_t tmp = strtol(str, pEnd, radix);
+#ifdef DARWIN
+  if (errno == EINVAL) errno = 0;
+#endif
 #ifdef TD_CHECK_STR_TO_INT_ERROR
   assert(errno != ERANGE);
   assert(errno != EINVAL);
@@ -362,6 +372,9 @@ uint32_t taosStr2UInt32(const char *str, char **pEnd, int32_t radix) {
 
 int16_t taosStr2Int16(const char *str, char **pEnd, int32_t radix) {
   int32_t tmp = strtol(str, pEnd, radix);
+#ifdef DARWIN
+  if (errno == EINVAL) errno = 0;
+#endif
 #ifdef TD_CHECK_STR_TO_INT_ERROR
   assert(errno != ERANGE);
   assert(errno != EINVAL);
@@ -373,6 +386,9 @@ int16_t taosStr2Int16(const char *str, char **pEnd, int32_t radix) {
 
 uint16_t taosStr2UInt16(const char *str, char **pEnd, int32_t radix) {
   uint32_t tmp = strtoul(str, pEnd, radix);
+#ifdef DARWIN
+  if (errno == EINVAL) errno = 0;
+#endif
 #ifdef TD_CHECK_STR_TO_INT_ERROR
   assert(errno != ERANGE);
   assert(errno != EINVAL);
@@ -394,6 +410,9 @@ int8_t taosStr2Int8(const char *str, char **pEnd, int32_t radix) {
 
 uint8_t taosStr2UInt8(const char *str, char **pEnd, int32_t radix) {
   uint32_t tmp = strtoul(str, pEnd, radix);
+#ifdef DARWIN
+  if (errno == EINVAL) errno = 0;
+#endif
 #ifdef TD_CHECK_STR_TO_INT_ERROR
   assert(errno != ERANGE);
   assert(errno != EINVAL);

source/os/src/osSysinfo.c

@@ -143,8 +143,11 @@ static int32_t taosGetSysCpuInfo(SysCpuInfo *cpuInfo) {
     cpuInfo->user = CompareFileTime(&pre_userTime, &userTime);
     cpuInfo->nice = 0;
   }
-#elif defined(_TD_DARWIN_64)
+#elif defined(DARWIN)
-  assert(0);
+  cpuInfo->idle = 0;
+  cpuInfo->system = 0;
+  cpuInfo->user = 0;
+  cpuInfo->nice = 0;
 #else
   TdFilePtr pFile = taosOpenFile(tsSysCpuFile, TD_FILE_READ | TD_FILE_STREAM);
   if (pFile == NULL) {
@@ -180,8 +183,11 @@ static int32_t taosGetProcCpuInfo(ProcCpuInfo *cpuInfo) {
     cpuInfo->cutime = 0;
     cpuInfo->cstime = 0;
 	}
-#elif defined(_TD_DARWIN_64)
+#elif defined(DARWIN)
-  assert(0);
+  cpuInfo->stime = 0;
+  cpuInfo->utime = 0;
+  cpuInfo->cutime = 0;
+  cpuInfo->cstime = 0;
 #else
   TdFilePtr pFile = taosOpenFile(tsProcCpuFile, TD_FILE_READ | TD_FILE_STREAM);
   if (pFile == NULL) {
@@ -359,7 +365,7 @@ int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores) {
   pCmd = taosOpenCmd("sysctl -n machdep.cpu.core_count");
   if (pCmd == NULL) return code;
   memset(buf, 0, sizeof(buf));
-  if (taosGetsCmd(pCmd, maxLen, cpuModel) > 0) {
+  if (taosGetsCmd(pCmd, sizeof(buf) - 1, buf) > 0) {
     code = 0;
     done |= 2;
     *numOfCores = atof(buf);

source/os/src/osTimezone.c

@@ -891,6 +891,7 @@ void taosGetSystemTimezone(char *outTimezoneStr, enum TdTimezone *tsTimezone) {
   time_t    tx1 = taosGetTimestampSec();
   struct tm tm1;
   taosLocalTime(&tx1, &tm1);
+  daylight = tm1.tm_isdst;
 
   /*
    * format example:

tests/pytest/util/dnodes.py

@@ -415,23 +415,15 @@ class TDDnode:
                     i += 1
                     if i > 50:
                         break
-                tailCmdStr = 'tail -f '
+                with open(logFile) as f:
-                popen = subprocess.Popen(
+                    timeout = time.time() + 60 * 2
-                    tailCmdStr + logFile,
+                    while True:
-                    stdout=subprocess.PIPE,
+                        line = f.readline().encode('utf-8')
-                    stderr=subprocess.PIPE,
+                        if bkey in line:
-                    shell=True)
+                            break
-                pid = popen.pid
+                        if time.time() > timeout:
-                # print('Popen.pid:' + str(pid))
+                            tdLog.exit('wait too long for taosd start')
-                timeout = time.time() + 60 * 2
+                    tdLog.debug("the dnode:%d has been started." % (self.index))
-                while True:
-                    line = popen.stdout.readline().strip()
-                    if bkey in line:
-                        popen.kill()
-                        break
-                    if time.time() > timeout:
-                        tdLog.exit('wait too long for taosd start')
-                tdLog.debug("the dnode:%d has been started." % (self.index))
             else:
                 tdLog.debug(
                     "wait 10 seconds for the dnode:%d to start." %
@@ -480,19 +472,21 @@ class TDDnode:
             toBeKilled = "valgrind.bin"
 
         if self.running != 0:
-            psCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}'" % toBeKilled
+            psCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}' | xargs" % toBeKilled
             processID = subprocess.check_output(
-                psCmd, shell=True).decode("utf-8")
+                psCmd, shell=True).decode("utf-8").strip()
 
             onlyKillOnceWindows = 0
             while(processID):
                 if not platform.system().lower() == 'windows' or (onlyKillOnceWindows == 0 and platform.system().lower() == 'windows'):
                     killCmd = "kill -INT %s > /dev/null 2>&1" % processID
+                    if platform.system().lower() == 'windows':
+                        killCmd = "kill -INT %s > nul 2>&1" % processID
                     os.system(killCmd)
                     onlyKillOnceWindows = 1
                 time.sleep(1)
                 processID = subprocess.check_output(
-                    psCmd, shell=True).decode("utf-8")
+                    psCmd, shell=True).decode("utf-8").strip()
             if not platform.system().lower() == 'windows':
                 for port in range(6030, 6041):
                     fuserCmd = "fuser -k -n tcp %d > /dev/null" % port
@@ -516,11 +510,11 @@ class TDDnode:
 
         if self.running != 0:
             if platform.system().lower() == 'windows':
-                psCmd = "for /f %%a in ('wmic process where \"name='taosd.exe' and CommandLine like '%%dnode%d%%'\" get processId ^| xargs echo ^| awk ^'{print $2}^' ^&^& echo aa') do @(ps | grep %%a | awk '{print $1}' )" % (self.index)
+                psCmd = "for /f %%a in ('wmic process where \"name='taosd.exe' and CommandLine like '%%dnode%d%%'\" get processId ^| xargs echo ^| awk ^'{print $2}^' ^&^& echo aa') do @(ps | grep %%a | awk '{print $1}' | xargs)" % (self.index)
             else:
-                psCmd = "ps -ef|grep -w %s| grep dnode%d|grep -v grep | awk '{print $2}'" % (toBeKilled,self.index)
+                psCmd = "ps -ef|grep -w %s| grep dnode%d|grep -v grep | awk '{print $2}' | xargs" % (toBeKilled,self.index)
             processID = subprocess.check_output(
-                psCmd, shell=True).decode("utf-8")
+                psCmd, shell=True).decode("utf-8").strip()
 
             onlyKillOnceWindows = 0
             while(processID):
@@ -530,7 +524,7 @@ class TDDnode:
                     onlyKillOnceWindows = 1
                 time.sleep(1)
                 processID = subprocess.check_output(
-                    psCmd, shell=True).decode("utf-8")
+                    psCmd, shell=True).decode("utf-8").strip()
             if self.valgrind:
                 time.sleep(2)
 
@@ -547,9 +541,9 @@ class TDDnode:
             toBeKilled = "valgrind.bin"
 
         if self.running != 0:
-            psCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}'" % toBeKilled
+            psCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}' | xargs" % toBeKilled
             processID = subprocess.check_output(
-                psCmd, shell=True).decode("utf-8")
+                psCmd, shell=True).decode("utf-8").strip()
 
             onlyKillOnceWindows = 0
             while(processID):
@@ -559,7 +553,7 @@ class TDDnode:
                     onlyKillOnceWindows = 1
                 time.sleep(1)
                 processID = subprocess.check_output(
-                    psCmd, shell=True).decode("utf-8")
+                    psCmd, shell=True).decode("utf-8").strip()
             for port in range(6030, 6041):
                 fuserCmd = "fuser -k -n tcp %d" % port
                 os.system(fuserCmd)
@@ -704,15 +698,15 @@ class TDDnodes:
         for i in range(len(self.dnodes)):
             self.dnodes[i].stop()
 
-        psCmd = "ps -ef | grep -w taosd | grep 'root' | grep -v grep| grep -v defunct | awk '{print $2}'"
+        psCmd = "ps -ef | grep -w taosd | grep 'root' | grep -v grep| grep -v defunct | awk '{print $2}' | xargs"
-        processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
+        processID = subprocess.check_output(psCmd, shell=True).decode("utf-8").strip()
         if processID:
             cmd = "sudo systemctl stop taosd"
             os.system(cmd)
         # if os.system(cmd) != 0 :
         # tdLog.exit(cmd)
-        psCmd = "ps -ef|grep -w taosd| grep -v grep| grep -v defunct | awk '{print $2}'"
+        psCmd = "ps -ef|grep -w taosd| grep -v grep| grep -v defunct | awk '{print $2}' | xargs"
-        processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
+        processID = subprocess.check_output(psCmd, shell=True).decode("utf-8").strip()
         while(processID):
             if platform.system().lower() == 'windows':
                 killCmd = "kill -9 %s > nul 2>&1" % processID
@@ -721,11 +715,11 @@ class TDDnodes:
             os.system(killCmd)
             time.sleep(1)
             processID = subprocess.check_output(
-                psCmd, shell=True).decode("utf-8")
+                psCmd, shell=True).decode("utf-8").strip()
 
         if self.killValgrind == 1:
-            psCmd = "ps -ef|grep -w valgrind.bin| grep -v grep | awk '{print $2}'"
+            psCmd = "ps -ef|grep -w valgrind.bin| grep -v grep | awk '{print $2}' | xargs"
-            processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
+            processID = subprocess.check_output(psCmd, shell=True).decode("utf-8").strip()
             while(processID):
                 if platform.system().lower() == 'windows':
                     killCmd = "kill -TERM %s > nul 2>&1" % processID
@@ -734,7 +728,7 @@ class TDDnodes:
                 os.system(killCmd)
                 time.sleep(1)
                 processID = subprocess.check_output(
-                    psCmd, shell=True).decode("utf-8")
+                    psCmd, shell=True).decode("utf-8").strip()
 
         # if os.system(cmd) != 0 :
         # tdLog.exit(cmd)