Merge remote-tracking branch 'origin/refact/tsdb_optimize' into refact/tsdb_optimize

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 9cb965f
+        GIT_TAG d2b5dec
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

docs/en/12-taos-sql/02-database.md

@@ -71,9 +71,9 @@ database_option: {
 - SINGLE_STABLE: specifies whether the database can contain more than one supertable.
   - 0: The database can contain multiple supertables.
   - 1: The database can contain only one supertable.
-- WAL_RETENTION_PERIOD: specifies the time after which WAL files are deleted. This parameter is used for data subscription. Enter a time in seconds. The default value is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted.
-- WAL_RETENTION_SIZE: specifies the size at which WAL files are deleted. This parameter is used for data subscription. Enter a size in KB. The default value is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted.
-- WAL_ROLL_PERIOD: specifies the time after which WAL files are rotated. After this period elapses, a new WAL file is created. The default value is 0. A value of 0 indicates that a new WAL file is created only after the previous WAL file was written to disk.
+- WAL_RETENTION_PERIOD: specifies the time after which WAL files are deleted. This parameter is used for data subscription. Enter a time in seconds. The default value of single copy is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted. The default value of multiple copy is 4 days.
+- WAL_RETENTION_SIZE: specifies the size at which WAL files are deleted. This parameter is used for data subscription. Enter a size in KB. The default value of single copy is 0. A value of 0 indicates that each WAL file is deleted immediately after its contents are written to disk. -1: WAL files are never deleted. The default value of multiple copy is -1.
+- WAL_ROLL_PERIOD: specifies the time after which WAL files are rotated. After this period elapses, a new WAL file is created. The default value of single copy is 0. A value of 0 indicates that a new WAL file is created only after the previous WAL file was written to disk. The default values of multiple copy is 1 day.
 - WAL_SEGMENT_SIZE: specifies the maximum size of a WAL file. After the current WAL file reaches this size, a new WAL file is created. The default value is 0. A value of 0 indicates that a new WAL file is created only after the previous WAL file was written to disk.
 
 ### Example Statement

docs/en/12-taos-sql/10-function.md

@@ -613,6 +613,7 @@ SELECT APERCENTILE(field_name, P[, algo_type]) FROM { tb_name | stb_name } [WHER
 **Explanations**：
 - _P_ is in range [0,100], when _P_ is 0, the result is same as using function MIN; when _P_ is 100, the result is same as function MAX.
 - `algo_type` can only be input as `default` or `t-digest` Enter `default` to use a histogram-based algorithm. Enter `t-digest` to use the t-digest algorithm to calculate the approximation of the quantile. `default` is used by default.
+- The approximation result of `t-digest` algorithm is sensitive to input data order. For example, when querying STable with different input data order there might be minor differences in calculated results.
 
 ### AVG
 

docs/en/12-taos-sql/24-show.md

@@ -5,16 +5,6 @@ title: SHOW Statement for Metadata
 
 `SHOW` command can be used to get brief system information. To get details about metatadata, information, and status in the system, please use `select` to query the tables in database `INFORMATION_SCHEMA`. 
 
-## SHOW ACCOUNTS
-
-```sql
-SHOW ACCOUNTS;
-```
-
-Shows information about tenants on the system.
-
-Note: TDengine Enterprise Edition only.
-
 ## SHOW APPS
 
 ```sql

docs/zh/08-connector/09-csharp.mdx

@@ -169,9 +169,9 @@ namespace TDengineExample
 
 ### 第三方驱动
 
-`Maikebing.Data.Taos` 是一个 TDengine 的 ADO.NET 连接器，支持 Linux，Windows 平台。该连接器由社区贡献者`麦壳饼@@maikebing` 提供，具体请参考:
+[`IoTSharp.Data.Taos`](https://github.com/IoTSharp/EntityFrameworkCore.Taos) 是一个 TDengine 的 ADO.NET 连接器,其中包含了用于EntityFrameworkCore 的提供程序 IoTSharp.EntityFrameworkCore.Taos 和健康检查组件 IoTSharp.HealthChecks.Taos ，支持 Linux，Windows 平台。该连接器由社区贡献者`麦壳饼@@maikebing` 提供，具体请参考:
 
-* 接口下载:<https://github.com/maikebing/Maikebing.EntityFrameworkCore.Taos>
+* 接口下载: <https://github.com/IoTSharp/EntityFrameworkCore.Taos>
 * 用法说明:<https://www.taosdata.com/blog/2020/11/02/1901.html>
 
 ## 常见问题

docs/zh/12-taos-sql/02-database.md

@@ -71,9 +71,9 @@ database_option: {
 - SINGLE_STABLE：表示此数据库中是否只可以创建一个超级表，用于超级表列非常多的情况。
   - 0：表示可以创建多张超级表。
   - 1：表示只可以创建一张超级表。
-- WAL_RETENTION_PERIOD：wal 文件的额外保留策略，用于数据订阅。wal 的保存时长，单位为 s。默认为 0，即落盘后立即删除。-1 表示不删除。
-- WAL_RETENTION_SIZE：wal 文件的额外保留策略，用于数据订阅。wal 的保存的最大上限，单位为 KB。默认为 0，即落盘后立即删除。-1 表示不删除。
-- WAL_ROLL_PERIOD：wal 文件切换时长，单位为 s。当 wal 文件创建并写入后，经过该时间，会自动创建一个新的 wal 文件。默认为 0，即仅在落盘时创建新文件。
+- WAL_RETENTION_PERIOD：wal 文件的额外保留策略，用于数据订阅。wal 的保存时长，单位为 s。单副本默认为 0，即落盘后立即删除。-1 表示不删除。多副本默认为 4 天。
+- WAL_RETENTION_SIZE：wal 文件的额外保留策略，用于数据订阅。wal 的保存的最大上限，单位为 KB。单副本默认为 0，即落盘后立即删除。多副本默认为-1，表示不删除。
+- WAL_ROLL_PERIOD：wal 文件切换时长，单位为 s。当 wal 文件创建并写入后，经过该时间，会自动创建一个新的 wal 文件。单副本默认为 0，即仅在落盘时创建新文件。多副本默认为 1 天。
 - WAL_SEGMENT_SIZE：wal 单个文件大小，单位为 KB。当前写入文件大小超过上限后会自动创建一个新的 wal 文件。默认为 0，即仅在落盘时创建新文件。
 
 ### 创建数据库示例

docs/zh/12-taos-sql/10-function.md

@@ -614,6 +614,7 @@ SELECT APERCENTILE(field_name, P[, algo_type]) FROM { tb_name | stb_name } [WHER
 **说明**：
 - P值范围是[0,100]，当为0时等同于MIN，为100时等同于MAX。
 - algo_type 取值为 "default" 或 "t-digest"。 输入为 "default" 时函数使用基于直方图算法进行计算。输入为 "t-digest" 时使用t-digest算法计算分位数的近似结果。如果不指定 algo_type 则使用 "default" 算法。
+- "t-digest"算法的近似结果对于输入数据顺序敏感，对超级表查询时不同的输入排序结果可能会有微小的误差。
 
 ### AVG
 

docs/zh/12-taos-sql/24-show.md

@@ -6,16 +6,6 @@ description: SHOW 命令的完整列表
 
 SHOW 命令可以用来获取简要的系统信息。若想获取系统中详细的各种元数据、系统信息和状态，请使用 select 语句查询 INFORMATION_SCHEMA 数据库中的表。
 
-## SHOW ACCOUNTS
-
-```sql
-SHOW ACCOUNTS;
-```
-
-显示当前系统中所有租户的信息。
-
-注：企业版独有
-
 ## SHOW APPS
 
 ```sql

docs/zh/14-reference/05-taosbenchmark.md

@@ -87,7 +87,7 @@ taosBenchmark -f <json file>
 <summary>subscribe.json</summary>
 
 ```json
-{{#include /taos-tools/example/tmq.json}}
+{{#include /taos-tools/example/subscribe.json}}
 ```
 
 </details>

include/common/tglobal.h

@@ -94,6 +94,7 @@ extern int64_t tsQueryBufferSizeBytes;   // maximum allowed usage buffer size in
 // query client
 extern int32_t tsQueryPolicy;
 extern int32_t tsQuerySmaOptimize;
+extern bool    tsQueryPlannerTrace;
 
 // client
 extern int32_t tsMinSlidingTime;
@@ -147,7 +148,7 @@ struct SConfig *taosGetCfg();
 void    taosSetAllDebugFlag(int32_t flag, bool rewrite);
 void    taosSetDebugFlag(int32_t *pFlagPtr, const char *flagName, int32_t flagVal, bool rewrite);
 int32_t taosSetCfg(SConfig *pCfg, char *name);
-void    taosLocalCfgForbiddenToChange(char* name, bool* forbidden);
+void    taosLocalCfgForbiddenToChange(char *name, bool *forbidden);
 
 #ifdef __cplusplus
 }

include/common/tmsg.h

@@ -1066,6 +1066,7 @@ typedef struct {
 typedef struct {
   int32_t vgId;
   int32_t syncState;
+  int64_t cacheUsage;
   int64_t numOfTables;
   int64_t numOfTimeSeries;
   int64_t totalStorage;
@@ -2994,6 +2995,7 @@ typedef struct {
 
 int32_t tEncodeSTaosxRsp(SEncoder* pEncoder, const STaosxRsp* pRsp);
 int32_t tDecodeSTaosxRsp(SDecoder* pDecoder, STaosxRsp* pRsp);
+void    tDeleteSTaosxRsp(STaosxRsp* pRsp);
 
 typedef struct {
   SMqRspHead head;

include/libs/nodes/cmdnodes.h

@@ -78,6 +78,9 @@ typedef struct SDatabaseOptions {
   int32_t     walRetentionSize;
   int32_t     walRollPeriod;
   int32_t     walSegmentSize;
+  bool        walRetentionPeriodIsSet;
+  bool        walRetentionSizeIsSet;
+  bool        walRollPeriodIsSet;
 } SDatabaseOptions;
 
 typedef struct SCreateDatabaseStmt {

include/libs/nodes/nodes.h

@@ -244,6 +244,7 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE,
   QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE,
   QUERY_NODE_PHYSICAL_PLAN_PARTITION,
+  QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION,
   QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC,
   QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC,
   QUERY_NODE_PHYSICAL_PLAN_DISPATCH,

include/libs/nodes/plannodes.h

@@ -488,6 +488,8 @@ typedef struct SPartitionPhysiNode {
   SNodeList* pTargets;
 } SPartitionPhysiNode;
 
+typedef SPartitionPhysiNode SStreamPartitionPhysiNode;
+
 typedef struct SDataSinkNode {
   ENodeType           type;
   SDataBlockDescNode* pInputDataBlockDesc;

include/os/osDir.h

@@ -56,6 +56,7 @@ void    taosRemoveDir(const char *dirname);
 bool    taosDirExist(const char *dirname);
 int32_t taosMkDir(const char *dirname);
 int32_t taosMulMkDir(const char *dirname);
+int32_t taosMulModeMkDir(const char *dirname, int mode);
 void    taosRemoveOldFiles(const char *dirname, int32_t keepDays);
 int32_t taosExpandDir(const char *dirname, char *outname, int32_t maxlen);
 int32_t taosRealPath(char *dirname, char *realPath, int32_t maxlen);

include/util/tdef.h

@@ -361,11 +361,14 @@ typedef enum ELogicConditionType {
 #define TSDB_DEFAULT_DB_SCHEMALESS      TSDB_DB_SCHEMALESS_OFF
 
 #define TSDB_DB_MIN_WAL_RETENTION_PERIOD -1
-#define TSDB_DEFAULT_DB_WAL_RETENTION_PERIOD (24 * 60 * 60 * 4)
+#define TSDB_REP_DEF_DB_WAL_RET_PERIOD   0
+#define TSDB_REPS_DEF_DB_WAL_RET_PERIOD  (24 * 60 * 60 * 4)
 #define TSDB_DB_MIN_WAL_RETENTION_SIZE   -1
-#define TSDB_DEFAULT_DB_WAL_RETENTION_SIZE   -1
+#define TSDB_REP_DEF_DB_WAL_RET_SIZE     0
+#define TSDB_REPS_DEF_DB_WAL_RET_SIZE    -1
 #define TSDB_DB_MIN_WAL_ROLL_PERIOD      0
-#define TSDB_DEFAULT_DB_WAL_ROLL_PERIOD      (24 * 60 * 60 * 1)
+#define TSDB_REP_DEF_DB_WAL_ROLL_PERIOD  0
+#define TSDB_REPS_DEF_DB_WAL_ROLL_PERIOD (24 * 60 * 60 * 1)
 #define TSDB_DB_MIN_WAL_SEGMENT_SIZE     0
 #define TSDB_DEFAULT_DB_WAL_SEGMENT_SIZE 0
 
@@ -386,7 +389,7 @@ typedef enum ELogicConditionType {
 
 #define TSDB_DEFAULT_EXPLAIN_VERBOSE false
 
-#define TSDB_EXPLAIN_RESULT_ROW_SIZE    (16*1024)
+#define TSDB_EXPLAIN_RESULT_ROW_SIZE    (16 * 1024)
 #define TSDB_EXPLAIN_RESULT_COLUMN_NAME "QUERY_PLAN"
 
 #define TSDB_MAX_FIELD_LEN             16384

packaging/deb/makedeb.sh

@@ -45,6 +45,7 @@ mkdir -p ${pkg_dir}${install_home_path}/include
 mkdir -p ${pkg_dir}${install_home_path}/script
 
 cp ${compile_dir}/../packaging/cfg/taos.cfg         ${pkg_dir}${install_home_path}/cfg
+cp ${compile_dir}/../packaging/cfg/taosd.service    ${pkg_dir}${install_home_path}/cfg
 if [ -f "${compile_dir}/test/cfg/taosadapter.toml" ]; then
     cp ${compile_dir}/test/cfg/taosadapter.toml		${pkg_dir}${install_home_path}/cfg || :
 fi

source/client/src/tmq.c

@@ -1773,7 +1773,7 @@ tmq_res_t tmq_get_res_type(TAOS_RES* res) {
     }
     return TMQ_RES_TABLE_META;
   } else if (TD_RES_TMQ_TAOSX(res)) {
-    return TMQ_RES_TAOSX;
+    return TMQ_RES_DATA;
   } else {
     return TMQ_RES_INVALID;
   }

source/common/src/systable.c

@@ -206,6 +206,7 @@ static const SSysDbTableSchema vgroupsSchema[] = {
     {.name = "v3_dnode", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "v3_status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
     {.name = "status", .bytes = 12 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "cacheload", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "nfiles", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "file_size", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "tsma", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT, .sysInfo = true},

source/common/src/tglobal.c

@@ -91,6 +91,7 @@ bool tsSmlDataFormat =
 // query
 int32_t tsQueryPolicy = 1;
 int32_t tsQuerySmaOptimize = 0;
+bool    tsQueryPlannerTrace = false;
 
 /*
  * denote if the server needs to compress response message at the application layer to client, including query rsp,
@@ -286,6 +287,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "compressColData", tsCompressColData, -1, 100000000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "queryPolicy", tsQueryPolicy, 1, 3, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "querySmaOptimize", tsQuerySmaOptimize, 0, 1, 1) != 0) return -1;
+  if (cfgAddBool(pCfg, "queryPlannerTrace", tsQueryPlannerTrace, true) != 0) return -1;
   if (cfgAddString(pCfg, "smlChildTableName", "", 1) != 0) return -1;
   if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, 1) != 0) return -1;
   if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, 1) != 0) return -1;
@@ -572,7 +574,6 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) {
   return 0;
 }
 
-
 static void taosSetClientLogCfg(SConfig *pCfg) {
   SConfigItem *pItem = cfgGetItem(pCfg, "logDir");
   tstrncpy(tsLogDir, cfgGetItem(pCfg, "logDir")->str, PATH_MAX);
@@ -643,6 +644,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
   tsNumOfTaskQueueThreads = cfgGetItem(pCfg, "numOfTaskQueueThreads")->i32;
   tsQueryPolicy = cfgGetItem(pCfg, "queryPolicy")->i32;
   tsQuerySmaOptimize = cfgGetItem(pCfg, "querySmaOptimize")->i32;
+  tsQueryPlannerTrace = cfgGetItem(pCfg, "queryPlannerTrace")->bval;
   return 0;
 }
 
@@ -972,6 +974,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
         tsQnodeShmSize = cfgGetItem(pCfg, "qnodeShmSize")->i32;
       } else if (strcasecmp("qDebugFlag", name) == 0) {
         qDebugFlag = cfgGetItem(pCfg, "qDebugFlag")->i32;
+      } else if (strcasecmp("queryPlannerTrace", name) == 0) {
+        tsQueryPlannerTrace = cfgGetItem(pCfg, "queryPlannerTrace")->bval;
       }
       break;
     }
@@ -1129,7 +1133,7 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
 
   taosSetAllDebugFlag(cfgGetItem(pCfg, "debugFlag")->i32, false);
 
-  if (taosMulMkDir(tsLogDir) != 0) {
+  if (taosMulModeMkDir(tsLogDir, 0777) != 0) {
     uError("failed to create dir:%s since %s", tsLogDir, terrstr());
     cfgCleanup(pCfg);
     return -1;
@@ -1241,16 +1245,14 @@ void taosCfgDynamicOptions(const char *option, const char *value) {
   }
 
   const char *options[] = {
-      "dDebugFlag",   "vDebugFlag",  "mDebugFlag",   "wDebugFlag",   "sDebugFlag",   "tsdbDebugFlag",
-      "tqDebugFlag",  "fsDebugFlag", "udfDebugFlag", "smaDebugFlag", "idxDebugFlag", "tdbDebugFlag",
-      "tmrDebugFlag", "uDebugFlag",  "smaDebugFlag", "rpcDebugFlag", "qDebugFlag",   "metaDebugFlag",
-      "jniDebugFlag",
+      "dDebugFlag",   "vDebugFlag",   "mDebugFlag",   "wDebugFlag",    "sDebugFlag",   "tsdbDebugFlag", "tqDebugFlag",
+      "fsDebugFlag",  "udfDebugFlag", "smaDebugFlag", "idxDebugFlag",  "tdbDebugFlag", "tmrDebugFlag",  "uDebugFlag",
+      "smaDebugFlag", "rpcDebugFlag", "qDebugFlag",   "metaDebugFlag", "jniDebugFlag",
   };
   int32_t *optionVars[] = {
-      &dDebugFlag,   &vDebugFlag,  &mDebugFlag,   &wDebugFlag,   &sDebugFlag,   &tsdbDebugFlag,
-      &tqDebugFlag,  &fsDebugFlag, &udfDebugFlag, &smaDebugFlag, &idxDebugFlag, &tdbDebugFlag,
-      &tmrDebugFlag, &uDebugFlag,  &smaDebugFlag, &rpcDebugFlag, &qDebugFlag,   &metaDebugFlag,
-      &jniDebugFlag,
+      &dDebugFlag,   &vDebugFlag,   &mDebugFlag,   &wDebugFlag,    &sDebugFlag,   &tsdbDebugFlag, &tqDebugFlag,
+      &fsDebugFlag,  &udfDebugFlag, &smaDebugFlag, &idxDebugFlag,  &tdbDebugFlag, &tmrDebugFlag,  &uDebugFlag,
+      &smaDebugFlag, &rpcDebugFlag, &qDebugFlag,   &metaDebugFlag, &jniDebugFlag,
   };
 
   int32_t optionSize = tListLen(options);

source/common/src/tmsg.c

@@ -994,6 +994,7 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
     SVnodeLoad *pload = taosArrayGet(pReq->pVloads, i);
     if (tEncodeI32(&encoder, pload->vgId) < 0) return -1;
     if (tEncodeI32(&encoder, pload->syncState) < 0) return -1;
+    if (tEncodeI64(&encoder, pload->cacheUsage) < 0) return -1;
     if (tEncodeI64(&encoder, pload->numOfTables) < 0) return -1;
     if (tEncodeI64(&encoder, pload->numOfTimeSeries) < 0) return -1;
     if (tEncodeI64(&encoder, pload->totalStorage) < 0) return -1;
@@ -1063,6 +1064,7 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
     SVnodeLoad vload = {0};
     if (tDecodeI32(&decoder, &vload.vgId) < 0) return -1;
     if (tDecodeI32(&decoder, &vload.syncState) < 0) return -1;
+    if (tDecodeI64(&decoder, &vload.cacheUsage) < 0) return -1;
     if (tDecodeI64(&decoder, &vload.numOfTables) < 0) return -1;
     if (tDecodeI64(&decoder, &vload.numOfTimeSeries) < 0) return -1;
     if (tDecodeI64(&decoder, &vload.totalStorage) < 0) return -1;
@@ -5984,6 +5986,17 @@ int32_t tDecodeSTaosxRsp(SDecoder *pDecoder, STaosxRsp *pRsp) {
   }
   return 0;
 }
+
+void tDeleteSTaosxRsp(STaosxRsp *pRsp) {
+  taosArrayDestroy(pRsp->blockDataLen);
+  taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree);
+  taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper);
+  taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree);
+
+  taosArrayDestroy(pRsp->createTableLen);
+  taosArrayDestroyP(pRsp->createTableReq, (FDelete)taosMemoryFree);
+}
+
 int32_t tEncodeSSingleDeleteReq(SEncoder *pEncoder, const SSingleDeleteReq *pReq) {
   if (tEncodeI64(pEncoder, pReq->uid) < 0) return -1;
   if (tEncodeI64(pEncoder, pReq->ts) < 0) return -1;

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

@@ -340,6 +340,7 @@ typedef struct {
   uint32_t  hashEnd;
   char      dbName[TSDB_DB_FNAME_LEN];
   int64_t   dbUid;
+  int64_t   cacheUsage;
   int64_t   numOfTables;
   int64_t   numOfTimeSeries;
   int64_t   totalStorage;

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

@@ -358,10 +358,10 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->numOfRetensions < 0) pCfg->numOfRetensions = 0;
   if (pCfg->schemaless < 0) pCfg->schemaless = TSDB_DB_SCHEMALESS_OFF;
   if (pCfg->walRetentionPeriod < 0 && pCfg->walRetentionPeriod != -1)
-    pCfg->walRetentionPeriod = TSDB_DEFAULT_DB_WAL_RETENTION_PERIOD;
+    pCfg->walRetentionPeriod = TSDB_REPS_DEF_DB_WAL_RET_PERIOD;
   if (pCfg->walRetentionSize < 0 && pCfg->walRetentionSize != -1)
-    pCfg->walRetentionSize = TSDB_DEFAULT_DB_WAL_RETENTION_SIZE;
-  if (pCfg->walRollPeriod < 0) pCfg->walRollPeriod = TSDB_DEFAULT_DB_WAL_ROLL_PERIOD;
+    pCfg->walRetentionSize = TSDB_REPS_DEF_DB_WAL_RET_SIZE;
+  if (pCfg->walRollPeriod < 0) pCfg->walRollPeriod = TSDB_REPS_DEF_DB_WAL_ROLL_PERIOD;
   if (pCfg->walSegmentSize < 0) pCfg->walSegmentSize = TSDB_DEFAULT_DB_WAL_SEGMENT_SIZE;
 }
 

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

@@ -347,6 +347,7 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
     SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVload->vgId);
     if (pVgroup != NULL) {
       if (pVload->syncState == TAOS_SYNC_STATE_LEADER) {
+        pVgroup->cacheUsage = pVload->cacheUsage;
         pVgroup->numOfTables = pVload->numOfTables;
         pVgroup->numOfTimeSeries = pVload->numOfTimeSeries;
         pVgroup->totalStorage = pVload->totalStorage;

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

@@ -693,6 +693,9 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppendNULL(pColInfo, numOfRows);
 
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->cacheUsage, false);
+
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppendNULL(pColInfo, numOfRows);
 

source/dnode/vnode/inc/vnode.h

@@ -155,6 +155,7 @@ int32_t tsdbGetTableSchema(SVnode *pVnode, int64_t uid, STSchema **pSchema, int6
 
 void   tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity);
 size_t tsdbCacheGetCapacity(SVnode *pVnode);
+size_t tsdbCacheGetUsage(SVnode *pVnode);
 
 // tq
 typedef struct SMetaTableInfo {

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

@@ -140,11 +140,12 @@ int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle);
 int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);
 
 // tqRead
-int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
+int32_t tqScan(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
+int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset);
 int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
 
 // tqExec
-int32_t tqLogScanExec(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, SMqDataRsp* pRsp);
+int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp);
 int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp);
 
 // tqMeta

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

@@ -196,6 +196,66 @@ int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, con
   return 0;
 }
 
+int32_t tqSendTaosxRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const STaosxRsp* pRsp) {
+  ASSERT(taosArrayGetSize(pRsp->blockData) == pRsp->blockNum);
+  ASSERT(taosArrayGetSize(pRsp->blockDataLen) == pRsp->blockNum);
+
+  if (pRsp->withSchema) {
+    ASSERT(taosArrayGetSize(pRsp->blockSchema) == pRsp->blockNum);
+  } else {
+    ASSERT(taosArrayGetSize(pRsp->blockSchema) == 0);
+  }
+
+  if (pRsp->reqOffset.type == TMQ_OFFSET__LOG) {
+    if (pRsp->blockNum > 0) {
+      ASSERT(pRsp->rspOffset.version > pRsp->reqOffset.version);
+    } else {
+      ASSERT(pRsp->rspOffset.version >= pRsp->reqOffset.version);
+    }
+  }
+
+  int32_t len = 0;
+  int32_t code = 0;
+  tEncodeSize(tEncodeSTaosxRsp, pRsp, len, code);
+  if (code < 0) {
+    return -1;
+  }
+  int32_t tlen = sizeof(SMqRspHead) + len;
+  void*   buf = rpcMallocCont(tlen);
+  if (buf == NULL) {
+    return -1;
+  }
+
+  ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__TAOSX_RSP;
+  ((SMqRspHead*)buf)->epoch = pReq->epoch;
+  ((SMqRspHead*)buf)->consumerId = pReq->consumerId;
+
+  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
+
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, abuf, len);
+  tEncodeSTaosxRsp(&encoder, pRsp);
+  tEncoderClear(&encoder);
+
+  SRpcMsg rsp = {
+      .info = pMsg->info,
+      .pCont = buf,
+      .contLen = tlen,
+      .code = 0,
+  };
+  tmsgSendRsp(&rsp);
+
+  char buf1[80] = {0};
+  char buf2[80] = {0};
+  tFormatOffset(buf1, 80, &pRsp->reqOffset);
+  tFormatOffset(buf2, 80, &pRsp->rspOffset);
+  tqDebug("taosx rsp, vgId:%d, from consumer:%" PRId64
+          ", (epoch %d) send rsp, block num: %d, reqOffset:%s, rspOffset:%s",
+          TD_VID(pTq->pVnode), pReq->consumerId, pReq->epoch, pRsp->blockNum, buf1, buf2);
+
+  return 0;
+}
+
 static FORCE_INLINE bool tqOffsetLessOrEqual(const STqOffset* pLeft, const STqOffset* pRight) {
   return pLeft->val.type == TMQ_OFFSET__LOG && pRight->val.type == TMQ_OFFSET__LOG &&
          pLeft->val.version <= pRight->val.version;
@@ -303,6 +363,22 @@ static int32_t tqInitDataRsp(SMqDataRsp* pRsp, const SMqPollReq* pReq, int8_t su
   return 0;
 }
 
+static int32_t tqInitTaosxRsp(STaosxRsp* pRsp, const SMqPollReq* pReq) {
+  pRsp->reqOffset = pReq->reqOffset;
+
+  pRsp->withTbName = 1;
+  pRsp->withSchema = 1;
+  pRsp->blockData = taosArrayInit(0, sizeof(void*));
+  pRsp->blockDataLen = taosArrayInit(0, sizeof(int32_t));
+  pRsp->blockTbName = taosArrayInit(0, sizeof(void*));
+  pRsp->blockSchema = taosArrayInit(0, sizeof(void*));
+
+  if (pRsp->blockData == NULL || pRsp->blockDataLen == NULL || pRsp->blockTbName == NULL || pRsp->blockSchema == NULL) {
+    return -1;
+  }
+  return 0;
+}
+
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   SMqPollReq*  pReq = pMsg->pCont;
   int64_t      consumerId = pReq->consumerId;
@@ -341,9 +417,6 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   tqDebug("tmq poll: consumer %" PRId64 " (epoch %d), subkey %s, recv poll req in vg %d, req offset %s", consumerId,
           pReq->epoch, pHandle->subKey, TD_VID(pTq->pVnode), buf);
 
-  SMqDataRsp dataRsp = {0};
-  tqInitDataRsp(&dataRsp, pReq, pHandle->execHandle.subType);
-
   // 2.reset offset if needed
   if (reqOffset.type > 0) {
     fetchOffsetNew = reqOffset;
@@ -367,6 +440,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
           tqOffsetResetToLog(&fetchOffsetNew, walGetFirstVer(pTq->pVnode->pWal));
         }
       } else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) {
+        SMqDataRsp dataRsp = {0};
+        tqInitDataRsp(&dataRsp, pReq, pHandle->execHandle.subType);
+
         tqOffsetResetToLog(&dataRsp.rspOffset, walGetLastVer(pTq->pVnode->pWal));
         tqDebug("tmq poll: consumer %" PRId64 ", subkey %s, vg %d, offset reset to %" PRId64, consumerId,
                 pHandle->subKey, TD_VID(pTq->pVnode), dataRsp.rspOffset.version);
@@ -380,16 +456,38 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
                 " in vg %d, subkey %s, reset none failed",
                 pHandle->subKey, consumerId, TD_VID(pTq->pVnode), pReq->subKey);
         terrno = TSDB_CODE_TQ_NO_COMMITTED_OFFSET;
+        return -1;
+      }
+    }
+  }
+
+  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+    SMqDataRsp dataRsp = {0};
+    tqInitDataRsp(&dataRsp, pReq, pHandle->execHandle.subType);
+    tqScanData(pTq, pHandle, &dataRsp, &fetchOffsetNew);
+
+    if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
       code = -1;
+    }
+
+    tqDebug("tmq poll: consumer %ld, subkey %s, vg %d, send data blockNum:%d, offset type:%d, uid:%ld, version:%ld",
+            consumerId, pHandle->subKey, TD_VID(pTq->pVnode), dataRsp.blockNum, dataRsp.rspOffset.type,
+            dataRsp.rspOffset.uid, dataRsp.rspOffset.version);
+
     tDeleteSMqDataRsp(&dataRsp);
     return code;
   }
-    }
-  }
 
-  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN || fetchOffsetNew.type != TMQ_OFFSET__LOG) {
+  // for taosx
+  ASSERT(pHandle->execHandle.subType != TOPIC_SUB_TYPE__COLUMN);
+
   SMqMetaRsp metaRsp = {0};
-    tqScan(pTq, pHandle, &dataRsp, &metaRsp, &fetchOffsetNew);
+
+  STaosxRsp taosxRsp = {0};
+  tqInitTaosxRsp(&taosxRsp, pReq);
+
+  if (fetchOffsetNew.type != TMQ_OFFSET__LOG) {
+    tqScan(pTq, pHandle, &taosxRsp, &metaRsp, &fetchOffsetNew);
 
     if (metaRsp.metaRspLen > 0) {
       if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) {
@@ -399,29 +497,30 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
               pHandle->subKey, TD_VID(pTq->pVnode), metaRsp.rspOffset.type, metaRsp.rspOffset.uid,
               metaRsp.rspOffset.version);
       taosMemoryFree(metaRsp.metaRsp);
-      goto OVER;
+      tDeleteSTaosxRsp(&taosxRsp);
+      return code;
     }
 
-    if (dataRsp.blockNum > 0) {
-      if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
+    if (taosxRsp.blockNum > 0) {
+      if (tqSendTaosxRsp(pTq, pMsg, pReq, &taosxRsp) < 0) {
         code = -1;
       }
-      goto OVER;
+      tDeleteSTaosxRsp(&taosxRsp);
+      return code;
     } else {
-      fetchOffsetNew = dataRsp.rspOffset;
+      fetchOffsetNew = taosxRsp.rspOffset;
     }
 
-    tqDebug("tmq poll: consumer %ld, subkey %s, vg %d, send data blockNum:%d, offset type:%d,uid:%ld,version:%ld",
-            consumerId, pHandle->subKey, TD_VID(pTq->pVnode), dataRsp.blockNum, dataRsp.rspOffset.type,
-            dataRsp.rspOffset.uid, dataRsp.rspOffset.version);
+    tqDebug("taosx poll: consumer %ld, subkey %s, vg %d, send data blockNum:%d, offset type:%d,uid:%ld,version:%ld",
+            consumerId, pHandle->subKey, TD_VID(pTq->pVnode), taosxRsp.blockNum, taosxRsp.rspOffset.type,
+            taosxRsp.rspOffset.uid, taosxRsp.rspOffset.version);
   }
 
-  if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__COLUMN && fetchOffsetNew.type == TMQ_OFFSET__LOG) {
+  if (fetchOffsetNew.type == TMQ_OFFSET__LOG) {
     int64_t fetchVer = fetchOffsetNew.version + 1;
     pCkHead = taosMemoryMalloc(sizeof(SWalCkHead) + 2048);
     if (pCkHead == NULL) {
-      code = -1;
-      goto OVER;
+      return -1;
     }
 
     walSetReaderCapacity(pHandle->pWalReader, 2048);
@@ -436,13 +535,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
       }
 
       if (tqFetchLog(pTq, pHandle, &fetchVer, &pCkHead) < 0) {
-        // TODO add push mgr
-
-        tqOffsetResetToLog(&dataRsp.rspOffset, fetchVer);
-        if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
+        tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer);
+        if (tqSendTaosxRsp(pTq, pMsg, pReq, &taosxRsp) < 0) {
           code = -1;
         }
-        goto OVER;
+        tDeleteSTaosxRsp(&taosxRsp);
+        if (pCkHead) taosMemoryFree(pCkHead);
+        return code;
       }
 
       SWalCont* pHead = &pCkHead->head;
@@ -453,17 +552,19 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
       if (pHead->msgType == TDMT_VND_SUBMIT) {
         SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
 
-        if (tqLogScanExec(pTq, pHandle, pCont, &dataRsp) < 0) {
+        if (tqTaosxScanLog(pTq, pHandle, pCont, &taosxRsp) < 0) {
           /*ASSERT(0);*/
         }
         // TODO batch optimization:
         // TODO continue scan until meeting batch requirement
-        if (dataRsp.blockNum > 0 /* threshold */) {
-          tqOffsetResetToLog(&dataRsp.rspOffset, fetchVer);
-          if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
+        if (taosxRsp.blockNum > 0 /* threshold */) {
+          tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer);
+          if (tqSendTaosxRsp(pTq, pMsg, pReq, &taosxRsp) < 0) {
             code = -1;
           }
-          goto OVER;
+          tDeleteSTaosxRsp(&taosxRsp);
+          if (pCkHead) taosMemoryFree(pCkHead);
+          return code;
         } else {
           fetchVer++;
         }
@@ -472,30 +573,22 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
         ASSERT(pHandle->fetchMeta);
         ASSERT(IS_META_MSG(pHead->msgType));
         tqDebug("fetch meta msg, ver:%" PRId64 ", type:%d", pHead->version, pHead->msgType);
-        SMqMetaRsp metaRsp = {0};
         tqOffsetResetToLog(&metaRsp.rspOffset, fetchVer);
         metaRsp.resMsgType = pHead->msgType;
         metaRsp.metaRspLen = pHead->bodyLen;
         metaRsp.metaRsp = pHead->body;
         if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) {
           code = -1;
-          goto OVER;
+          taosMemoryFree(pCkHead);
+          return code;
         }
         code = 0;
-        goto OVER;
-      }
+        if (pCkHead) taosMemoryFree(pCkHead);
+        return code;
       }
     }
-
-  // send empty to client
-  if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
-    code = -1;
   }
-
-OVER:
-  if (pCkHead) taosMemoryFree(pCkHead);
-  tDeleteSMqDataRsp(&dataRsp);
-  return code;
+  return 0;
 }
 
 int32_t tqProcessVgDeleteReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
@@ -602,10 +695,10 @@ int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLe
       pHandle->execHandle.pExecReader = tqOpenReader(pTq->pVnode);
       pHandle->execHandle.execDb.pFilterOutTbUid =
           taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
-      buildSnapContext(handle.meta, handle.version, 0, pHandle->execHandle.subType, pHandle->fetchMeta, (SSnapContext **)(&handle.sContext));
+      buildSnapContext(handle.meta, handle.version, 0, pHandle->execHandle.subType, pHandle->fetchMeta,
+                       (SSnapContext**)(&handle.sContext));
 
-      pHandle->execHandle.task =
-          qCreateQueueExecTaskInfo(NULL, &handle, NULL, NULL);
+      pHandle->execHandle.task = qCreateQueueExecTaskInfo(NULL, &handle, NULL, NULL);
     } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
       pHandle->pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
 

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

@@ -60,7 +60,7 @@ static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp) {
   return 0;
 }
 
-int64_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset) {
+int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset) {
   const STqExecHandle* pExec = &pHandle->execHandle;
   ASSERT(pExec->subType == TOPIC_SUB_TYPE__COLUMN);
 
@@ -89,18 +89,41 @@ int64_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
     if (qExecTask(task, &pDataBlock, &ts) < 0) {
       ASSERT(0);
     }
-    tqDebug("tmq task execute end, get %p", pDataBlock);
+    tqDebug("tmq task executed, get %p", pDataBlock);
+
+    if (pDataBlock == NULL) {
+      break;
+    }
 
-    if (pDataBlock) {
     tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols);
     pRsp->blockNum++;
+
+    if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
+      rowCnt += pDataBlock->info.rows;
+      if (rowCnt >= 4096) break;
     }
   }
 
+  if (qStreamExtractOffset(task, &pRsp->rspOffset) < 0) {
+    ASSERT(0);
+    return -1;
+  }
+  ASSERT(pRsp->rspOffset.type != 0);
+
+  if (pRsp->withTbName) {
+    if (pRsp->rspOffset.type == TMQ_OFFSET__LOG) {
+      int64_t uid = pExec->pExecReader->msgIter.uid;
+      tqAddTbNameToRsp(pTq, uid, pRsp);
+    } else {
+      pRsp->withTbName = false;
+    }
+  }
+  ASSERT(pRsp->withSchema == false);
+
   return 0;
 }
 
-int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
+int32_t tqScan(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
   const STqExecHandle* pExec = &pHandle->execHandle;
   qTaskInfo_t          task = pExec->task;
 
@@ -134,7 +157,7 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
         int64_t uid = 0;
         if (pOffset->type == TMQ_OFFSET__LOG) {
           uid = pExec->pExecReader->msgIter.uid;
-          if (tqAddTbNameToRsp(pTq, uid, pRsp) < 0) {
+          if (tqAddTbNameToRsp(pTq, uid, (SMqDataRsp*)pRsp) < 0) {
             continue;
           }
         } else {
@@ -144,18 +167,14 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
       }
       if (pRsp->withSchema) {
         if (pOffset->type == TMQ_OFFSET__LOG) {
-          tqAddBlockSchemaToRsp(pExec, pRsp);
+          tqAddBlockSchemaToRsp(pExec, (SMqDataRsp*)pRsp);
         } else {
           SSchemaWrapper* pSW = tCloneSSchemaWrapper(qExtractSchemaFromTask(task));
           taosArrayPush(pRsp->blockSchema, &pSW);
         }
       }
 
-      if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-        tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols);
-      } else {
-        tqAddBlockDataToRsp(pDataBlock, pRsp, taosArrayGetSize(pDataBlock->pDataBlock));
-      }
+      tqAddBlockDataToRsp(pDataBlock, (SMqDataRsp*)pRsp, taosArrayGetSize(pDataBlock->pDataBlock));
       pRsp->blockNum++;
       if (pOffset->type == TMQ_OFFSET__LOG) {
         continue;
@@ -165,7 +184,6 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
       }
     }
 
-    if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__COLUMN) {
     if (pDataBlock == NULL && pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
       if (qStreamExtractPrepareUid(task) != 0) {
         continue;
@@ -191,7 +209,6 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
 
     *pMetaRsp = *tmp;
     tqDebug("tmqsnap task exec exited, get meta");
-    }
 
     tqDebug("task exec exited");
     break;
@@ -205,12 +222,11 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
   return 0;
 }
 
-int32_t tqLogScanExec(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, SMqDataRsp* pRsp) {
+int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp) {
   STqExecHandle* pExec = &pHandle->execHandle;
   ASSERT(pExec->subType != TOPIC_SUB_TYPE__COLUMN);
 
   if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
-    pRsp->withSchema = 1;
     STqReader* pReader = pExec->pExecReader;
     tqReaderSetDataMsg(pReader, pReq, 0);
     while (tqNextDataBlock(pReader)) {
@@ -220,18 +236,17 @@ int32_t tqLogScanExec(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, SMqDataRsp
       }
       if (pRsp->withTbName) {
         int64_t uid = pExec->pExecReader->msgIter.uid;
-        if (tqAddTbNameToRsp(pTq, uid, pRsp) < 0) {
+        if (tqAddTbNameToRsp(pTq, uid, (SMqDataRsp*)pRsp) < 0) {
           blockDataFreeRes(&block);
           continue;
         }
       }
-      tqAddBlockDataToRsp(&block, pRsp, taosArrayGetSize(block.pDataBlock));
+      tqAddBlockDataToRsp(&block, (SMqDataRsp*)pRsp, taosArrayGetSize(block.pDataBlock));
       blockDataFreeRes(&block);
-      tqAddBlockSchemaToRsp(pExec, pRsp);
+      tqAddBlockSchemaToRsp(pExec, (SMqDataRsp*)pRsp);
       pRsp->blockNum++;
     }
   } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
-    pRsp->withSchema = 1;
     STqReader* pReader = pExec->pExecReader;
     tqReaderSetDataMsg(pReader, pReq, 0);
     while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {
@@ -241,24 +256,25 @@ int32_t tqLogScanExec(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, SMqDataRsp
       }
       if (pRsp->withTbName) {
         int64_t uid = pExec->pExecReader->msgIter.uid;
-        if (tqAddTbNameToRsp(pTq, uid, pRsp) < 0) {
+        if (tqAddTbNameToRsp(pTq, uid, (SMqDataRsp*)pRsp) < 0) {
           blockDataFreeRes(&block);
           continue;
         }
       }
-      tqAddBlockDataToRsp(&block, pRsp, taosArrayGetSize(block.pDataBlock));
+      tqAddBlockDataToRsp(&block, (SMqDataRsp*)pRsp, taosArrayGetSize(block.pDataBlock));
       blockDataFreeRes(&block);
-      tqAddBlockSchemaToRsp(pExec, pRsp);
+      tqAddBlockSchemaToRsp(pExec, (SMqDataRsp*)pRsp);
       pRsp->blockNum++;
     }
-#if 0
+#if 1
     if (pHandle->fetchMeta && pRsp->blockNum) {
       SSubmitMsgIter iter = {0};
       tInitSubmitMsgIter(pReq, &iter);
       STaosxRsp* pXrsp = (STaosxRsp*)pRsp;
       while (1) {
         SSubmitBlk* pBlk = NULL;
-        if (tGetSubmitMsgNext(&iter, &pBlk) < 0) return -1;
+        if (tGetSubmitMsgNext(&iter, &pBlk) < 0) break;
+        if (pBlk == NULL) break;
         if (pBlk->schemaLen > 0) {
           if (pXrsp->createTableNum == 0) {
             pXrsp->createTableLen = taosArrayInit(0, sizeof(int32_t));

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

@@ -1334,3 +1334,5 @@ void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) {
 }
 
 size_t tsdbCacheGetCapacity(SVnode *pVnode) { return taosLRUCacheGetCapacity(pVnode->pTsdb->lruCache); }
+
+size_t tsdbCacheGetUsage(SVnode *pVnode) { return taosLRUCacheGetUsage(pVnode->pTsdb->lruCache); }

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

@@ -2055,7 +2055,7 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol
   size += pBlockCol->szOffset;
 
   // value
-  if (pColData->flag != (HAS_NULL | HAS_NONE)) {
+  if ((pColData->flag != (HAS_NULL | HAS_NONE)) && pColData->nData) {
     code = tsdbCmprData((uint8_t *)pColData->pData, pColData->nData, pColData->type, cmprAlg, ppOut, nOut + size,
                         &pBlockCol->szValue, ppBuf);
     if (code) goto _exit;

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

@@ -368,6 +368,7 @@ _exit:
 int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
   pLoad->vgId = TD_VID(pVnode);
   pLoad->syncState = syncGetMyRole(pVnode->sync);
+  pLoad->cacheUsage = tsdbCacheGetUsage(pVnode);
   pLoad->numOfTables = metaGetTbNum(pVnode->pMeta);
   pLoad->numOfTimeSeries = metaGetTimeSeriesNum(pVnode->pMeta);
   pLoad->totalStorage = (int64_t)3 * 1073741824;
@@ -424,8 +425,8 @@ int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t vnodeGetStbIdList(SVnode* pVnode, int64_t suid, SArray* list) {
-  SMStbCursor* pCur = metaOpenStbCursor(pVnode->pMeta, suid);
+int32_t vnodeGetStbIdList(SVnode *pVnode, int64_t suid, SArray *list) {
+  SMStbCursor *pCur = metaOpenStbCursor(pVnode->pMeta, suid);
   if (!pCur) {
     return TSDB_CODE_FAILED;
   }

source/libs/executor/inc/executorimpl.h

@@ -143,6 +143,7 @@ typedef struct {
   STqOffsetVal   prepareStatus; // for tmq
   STqOffsetVal   lastStatus;    // for tmq
   SMqMetaRsp     metaRsp;       // for tmq fetching meta
+  int8_t  returned;
   int64_t snapshotVer;
 
   SSchemaWrapper *schema;
@@ -409,6 +410,7 @@ typedef enum EStreamScanMode {
   STREAM_SCAN_FROM_READERHANDLE = 1,
   STREAM_SCAN_FROM_RES,
   STREAM_SCAN_FROM_UPDATERES,
+  STREAM_SCAN_FROM_DELETERES,
   STREAM_SCAN_FROM_DATAREADER_RETRIEVE,
   STREAM_SCAN_FROM_DATAREADER_RANGE,
 } EStreamScanMode;
@@ -437,12 +439,24 @@ typedef struct SStreamAggSupporter {
   SSDataBlock*   pScanBlock;
 } SStreamAggSupporter;
 
-typedef struct SessionWindowSupporter {
+typedef struct SWindowSupporter {
   SStreamAggSupporter* pStreamAggSup;
   int64_t              gap;
   uint16_t             parentType;
   SAggSupporter*       pIntervalAggSup;
-} SessionWindowSupporter;
+} SWindowSupporter;
+
+typedef struct SPartitionBySupporter {
+  SArray*         pGroupCols;     // group by columns, SArray<SColumn>
+  SArray*         pGroupColVals;  // current group column values, SArray<SGroupKeys>
+  char*           keyBuf;         // group by keys for hash
+  bool            needCalc;       // partition by column
+} SPartitionBySupporter;
+
+typedef struct SPartitionDataInfo {
+  uint64_t        groupId;
+  SArray*         rowIds;
+} SPartitionDataInfo;
 
 typedef struct STimeWindowSupp {
   int8_t           calTrigger;
@@ -477,7 +491,9 @@ typedef struct SStreamScanInfo {
   SOperatorInfo*         pStreamScanOp;
   SOperatorInfo*         pTableScanOp;
   SArray*                childIds;
-  SessionWindowSupporter sessionSup;
+  SWindowSupporter       windowSup;
+  SPartitionBySupporter  partitionSup;
+  SExprSupp*             pPartScalarSup;
   bool                   assignBlockUid;  // assign block uid to groupId, temporarily used for generating rollup SMA.
   int32_t                scanWinIndex;    // for state operator
   int32_t                pullDataResIndex;
@@ -690,7 +706,6 @@ typedef struct SPartitionOperatorInfo {
   SArray*        sortedGroupArray;   // SDataGroupInfo sorted by group id
   int32_t        groupIndex;  // group index
   int32_t        pageIndex;   // page index of current group
-  SSDataBlock*   pUpdateRes;
   SExprSupp      scalarSup;
 } SPartitionOperatorInfo;
 
@@ -742,8 +757,8 @@ typedef struct SStreamSessionAggOperatorInfo {
   SSDataBlock*         pWinBlock;       // window result
   SqlFunctionCtx*      pDummyCtx;       // for combine
   SSDataBlock*         pDelRes;         // delete result
-  bool                 returnDelete;
   SSDataBlock*         pUpdateRes;      // update window
+  bool                 returnUpdate;
   SHashObj*            pStDeleted;
   void*                pDelIterator;
   SArray*              pChildren;       // cache for children's result; final stream operator
@@ -752,6 +767,16 @@ typedef struct SStreamSessionAggOperatorInfo {
   bool                 ignoreExpiredData;
 } SStreamSessionAggOperatorInfo;
 
+typedef struct SStreamPartitionOperatorInfo {
+  SOptrBasicInfo        binfo;
+  SPartitionBySupporter partitionSup;
+  SExprSupp             scalarSup;
+  SHashObj*             pPartitions;
+  void*                 parIte;
+  SSDataBlock*          pInputDataBlock;
+  int32_t               tsColIndex;
+} SStreamPartitionOperatorInfo;
+
 typedef struct STimeSliceOperatorInfo {
   SSDataBlock*            pRes;
   STimeWindow             win;
@@ -953,6 +978,9 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
+
+SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
+
 SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SSortMergeJoinPhysiNode* pJoinNode,
                                            SExecTaskInfo* pTaskInfo);
@@ -1021,8 +1049,9 @@ bool functionNeedToExecute(SqlFunctionCtx* pCtx);
 bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
 bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
 bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
-void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, int32_t uidCol, uint64_t* pID);
+void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, uint64_t* pGp);
 void printDataBlock(SSDataBlock* pBlock, const char* flag);
+uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
 
 int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition,
                                        SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,

source/libs/executor/src/executor.c

@@ -745,6 +745,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
   SOperatorInfo* pOperator = pTaskInfo->pRoot;
   ASSERT(pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE);
   pTaskInfo->streamInfo.prepareStatus = *pOffset;
+  pTaskInfo->streamInfo.returned = 0;
   if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus)) {
     return 0;
   }

source/libs/executor/src/executorimpl.c

@@ -4150,6 +4150,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     pOptr = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
   } 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);
   } 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

@@ -830,3 +830,205 @@ int32_t setGroupResultOutputBuf(SOperatorInfo* pOperator, SOptrBasicInfo* binfo,
   setResultRowInitCtx(pResultRow, pCtx, numOfCols, pOperator->exprSupp.rowEntryInfoOffset);
   return TSDB_CODE_SUCCESS;
 }
+
+uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId) {
+  if (pExprSup->pExprInfo != NULL) {
+    int32_t code = projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
+    if (code != TSDB_CODE_SUCCESS) {
+      qError("calaculate group id error, code:%d", code);
+    }
+  }
+  recordNewGroupKeys(pParSup->pGroupCols, pParSup->pGroupColVals, pBlock, rowId);
+  int32_t len = buildGroupKeys(pParSup->keyBuf, pParSup->pGroupColVals);
+  uint64_t groupId = calcGroupId(pParSup->keyBuf, len);
+  return groupId;
+}
+
+static bool hasRemainPartion(SStreamPartitionOperatorInfo* pInfo) {
+  return pInfo->parIte != NULL;
+}
+
+static SSDataBlock* buildStreamPartitionResult(SOperatorInfo* pOperator) {
+  SStreamPartitionOperatorInfo* pInfo = pOperator->info;
+  SSDataBlock* pDest = pInfo->binfo.pRes;
+  ASSERT(hasRemainPartion(pInfo));
+  SPartitionDataInfo* pParInfo = (SPartitionDataInfo*)pInfo->parIte;
+  blockDataCleanup(pDest);
+  int32_t rows = taosArrayGetSize(pParInfo->rowIds);
+  SSDataBlock* pSrc = pInfo->pInputDataBlock;
+  for (int32_t i = 0; i < rows; i++) {
+    int32_t rowIndex = *(int32_t*)taosArrayGet(pParInfo->rowIds, i);
+    for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; j++) {
+      int32_t slotId = pOperator->exprSupp.pExprInfo[j].base.pParam[0].pCol->slotId;
+      SColumnInfoData* pSrcCol = taosArrayGet(pSrc->pDataBlock, slotId);
+      SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, j);
+      bool isNull = colDataIsNull(pSrcCol, pSrc->info.rows, rowIndex, NULL);
+      char* pSrcData = colDataGetData(pSrcCol, rowIndex);
+      colDataAppend(pDestCol, pDest->info.rows, pSrcData, isNull);
+    }
+    pDest->info.rows++;
+  }
+  blockDataUpdateTsWindow(pDest, pInfo->tsColIndex);
+  pDest->info.groupId = pParInfo->groupId;
+  pOperator->resultInfo.totalRows += pDest->info.rows;
+  pInfo->parIte = taosHashIterate(pInfo->pPartitions, pInfo->parIte);
+  ASSERT(pDest->info.rows > 0);
+  printDataBlock(pDest, "stream partitionby");
+  return pDest;
+}
+
+static void doStreamHashPartitionImpl(SStreamPartitionOperatorInfo* pInfo, SSDataBlock* pBlock) {
+  pInfo->pInputDataBlock = pBlock;
+  for (int32_t i = 0; i < pBlock->info.rows; ++i) {
+    recordNewGroupKeys(pInfo->partitionSup.pGroupCols, pInfo->partitionSup.pGroupColVals, pBlock, i);
+    int32_t keyLen = buildGroupKeys(pInfo->partitionSup.keyBuf, pInfo->partitionSup.pGroupColVals);
+    SPartitionDataInfo* pParData = 
+        (SPartitionDataInfo*) taosHashGet(pInfo->pPartitions, pInfo->partitionSup.keyBuf, keyLen);
+    if (pParData) {
+      taosArrayPush(pParData->rowIds, &i);
+    } else {
+      SPartitionDataInfo newParData = {0};
+      newParData.groupId = calcGroupId(pInfo->partitionSup.keyBuf, keyLen);
+      newParData.rowIds = taosArrayInit(64, sizeof(int32_t));
+      taosArrayPush(newParData.rowIds, &i);
+      taosHashPut(pInfo->pPartitions, pInfo->partitionSup.keyBuf, keyLen, &newParData,
+          sizeof(SPartitionDataInfo));
+    }
+  }
+}
+
+static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) {
+  if (pOperator->status == OP_EXEC_DONE) {
+    return NULL;
+  }
+
+  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
+  SStreamPartitionOperatorInfo* pInfo = pOperator->info;
+  if (hasRemainPartion(pInfo)) {
+    return buildStreamPartitionResult(pOperator);
+  }
+
+  int64_t st = taosGetTimestampUs();
+  SOperatorInfo* downstream = pOperator->pDownstream[0];
+  {
+    pInfo->pInputDataBlock = NULL;
+    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
+    if (pBlock == NULL) {
+      doSetOperatorCompleted(pOperator);
+      return NULL;
+    }
+    printDataBlock(pBlock, "stream partitionby recv");
+    switch (pBlock->info.type) {
+      case STREAM_NORMAL:
+      case STREAM_PULL_DATA:
+      case STREAM_INVALID:
+        pInfo->binfo.pRes->info.type = pBlock->info.type;
+        break;
+      default:
+        return pBlock;
+    }
+
+    // there is an scalar expression that needs to be calculated right before apply the group aggregation.
+    if (pInfo->scalarSup.pExprInfo != NULL) {
+      pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock,
+          pInfo->scalarSup.pCtx, pInfo->scalarSup.numOfExprs, NULL);
+      if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
+        longjmp(pTaskInfo->env, pTaskInfo->code);
+      }
+    }
+    taosHashClear(pInfo->pPartitions);
+    doStreamHashPartitionImpl(pInfo, pBlock);
+  }
+  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
+  
+  pInfo->parIte = taosHashIterate(pInfo->pPartitions, NULL);
+  return buildStreamPartitionResult(pOperator);
+}
+
+static void destroyStreamPartitionOperatorInfo(void* param) {
+  SStreamPartitionOperatorInfo* pInfo = (SStreamPartitionOperatorInfo*)param;
+  cleanupBasicInfo(&pInfo->binfo);
+  taosArrayDestroy(pInfo->partitionSup.pGroupCols);
+
+  for(int i = 0; i < taosArrayGetSize(pInfo->partitionSup.pGroupColVals); i++){
+    SGroupKeys key = *(SGroupKeys*)taosArrayGet(pInfo->partitionSup.pGroupColVals, i);
+    taosMemoryFree(key.pData);
+  }
+  taosArrayDestroy(pInfo->partitionSup.pGroupColVals);
+
+  taosMemoryFree(pInfo->partitionSup.keyBuf);
+  cleanupExprSupp(&pInfo->scalarSup);
+  taosMemoryFreeClear(param);
+}
+
+void initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup, SExprSupp* pExpr) {
+  if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
+    return;
+  }
+  SStreamScanInfo* pScanInfo = downstream->info;
+  pScanInfo->partitionSup = *pParSup;
+  pScanInfo->pPartScalarSup = pExpr;
+}
+
+SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo) {
+  SStreamPartitionOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamPartitionOperatorInfo));
+  SOperatorInfo*                pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
+  if (pInfo == NULL || pOperator == NULL) {
+    goto _error;
+  }
+  int32_t code = TSDB_CODE_SUCCESS;
+  pInfo->partitionSup.pGroupCols = extractPartitionColInfo(pPartNode->pPartitionKeys);
+
+  if (pPartNode->pExprs != NULL) {
+    int32_t num = 0;
+    SExprInfo* pCalExprInfo = createExprInfo(pPartNode->pExprs, NULL, &num);
+    code = initExprSupp(&pInfo->scalarSup, pCalExprInfo, num);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _error;
+    }
+  }
+
+  int32_t keyLen = 0;
+  code = initGroupOptrInfo(&pInfo->partitionSup.pGroupColVals, &keyLen, &pInfo->partitionSup.keyBuf, pInfo->partitionSup.pGroupCols);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _error;
+  }
+  pInfo->partitionSup.needCalc = true;
+
+  SSDataBlock* pResBlock = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);
+  if (!pResBlock) {
+    goto _error;
+  }
+  blockDataEnsureCapacity(pResBlock, 4096);
+  pInfo->binfo.pRes       = pResBlock;
+  pInfo->parIte           = NULL;
+  pInfo->pInputDataBlock  = NULL;
+  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
+  pInfo->pPartitions      = taosHashInit(1024, hashFn, false, HASH_NO_LOCK);
+  pInfo->tsColIndex       = 0;
+
+  int32_t numOfCols = 0;
+  SExprInfo* pExprInfo = createExprInfo(pPartNode->pTargets, NULL, &numOfCols);
+
+  pOperator->name         = "StreamPartitionOperator";
+  pOperator->blocking     = false;
+  pOperator->status       = OP_NOT_OPENED;
+  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION;
+  pOperator->exprSupp.numOfExprs   = numOfCols;
+  pOperator->exprSupp.pExprInfo    = pExprInfo;
+  pOperator->info         = pInfo;
+  pOperator->pTaskInfo    = pTaskInfo;
+  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamHashPartition, NULL, NULL, destroyStreamPartitionOperatorInfo,
+                                         NULL, NULL, NULL);
+
+  initParDownStream(downstream, &pInfo->partitionSup, &pInfo->scalarSup);
+  code = appendDownstream(pOperator, &downstream, 1);
+  return pOperator;
+
+  _error:
+  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
+  taosMemoryFreeClear(pInfo);
+  taosMemoryFreeClear(pOperator);
+  return NULL;
+}
+

source/libs/monitor/src/monMsg.c

@@ -510,6 +510,7 @@ int32_t tSerializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInfo)
     SVnodeLoad *pLoad = taosArrayGet(pInfo->pVloads, i);
     if (tEncodeI32(&encoder, pLoad->vgId) < 0) return -1;
     if (tEncodeI32(&encoder, pLoad->syncState) < 0) return -1;
+    if (tEncodeI64(&encoder, pLoad->cacheUsage) < 0) return -1;
     if (tEncodeI64(&encoder, pLoad->numOfTables) < 0) return -1;
     if (tEncodeI64(&encoder, pLoad->numOfTimeSeries) < 0) return -1;
     if (tEncodeI64(&encoder, pLoad->totalStorage) < 0) return -1;
@@ -544,6 +545,7 @@ int32_t tDeserializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInf
     SVnodeLoad load = {0};
     if (tDecodeI32(&decoder, &load.vgId) < 0) return -1;
     if (tDecodeI32(&decoder, &load.syncState) < 0) return -1;
+    if (tDecodeI64(&decoder, &load.cacheUsage) < 0) return -1;
     if (tDecodeI64(&decoder, &load.numOfTables) < 0) return -1;
     if (tDecodeI64(&decoder, &load.numOfTimeSeries) < 0) return -1;
     if (tDecodeI64(&decoder, &load.totalStorage) < 0) return -1;
@@ -594,7 +596,6 @@ int32_t tDeserializeSMonMloadInfo(void *buf, int32_t bufLen, SMonMloadInfo *pInf
   return 0;
 }
 
-
 int32_t tSerializeSQnodeLoad(void *buf, int32_t bufLen, SQnodeLoad *pInfo) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
@@ -639,5 +640,3 @@ int32_t tDeserializeSQnodeLoad(void *buf, int32_t bufLen, SQnodeLoad *pInfo) {
   tDecoderClear(&decoder);
   return 0;
 }
-
-

source/libs/nodes/src/nodesCloneFuncs.c

@@ -777,6 +777,7 @@ SNode* nodesCloneNode(const SNode* pNode) {
       code = physiSessionCopy((const SSessionWinodwPhysiNode*)pNode, (SSessionWinodwPhysiNode*)pDst);
       break;
     case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
       code = physiPartitionCopy((const SPartitionPhysiNode*)pNode, (SPartitionPhysiNode*)pDst);
       break;
     default:

source/libs/nodes/src/nodesCodeFuncs.c

@@ -265,6 +265,8 @@ const char* nodesNodeName(ENodeType type) {
       return "PhysiStreamStateWindow";
     case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
       return "PhysiPartition";
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
+      return "PhysiStreamPartition";
     case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC:
       return "PhysiIndefRowsFunc";
     case QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC:
@@ -4485,6 +4487,7 @@ 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_INDEF_ROWS_FUNC:
       return physiIndefRowsFuncNodeToJson(pObj, pJson);
@@ -4632,6 +4635,7 @@ 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_INDEF_ROWS_FUNC:
       return jsonToPhysiIndefRowsFuncNode(pJson, pObj);

source/libs/nodes/src/nodesTraverseFuncs.c

@@ -537,7 +537,8 @@ static EDealRes dispatchPhysiPlan(SNode* pNode, ETraversalOrder order, FNodeWalk
       }
       break;
     }
-    case QUERY_NODE_PHYSICAL_PLAN_PARTITION: {
+    case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION: {
       SPartitionPhysiNode* pPart = (SPartitionPhysiNode*)pNode;
       res = walkPhysiNode((SPhysiNode*)pNode, order, walker, pContext);
       if (DEAL_RES_ERROR != res && DEAL_RES_END != res) {

source/libs/nodes/src/nodesUtilFuncs.c

@@ -322,6 +322,8 @@ SNode* nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SStreamStateWinodwPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
       return makeNode(type, sizeof(SPartitionPhysiNode));
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:
+      return makeNode(type, sizeof(SStreamPartitionPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC:
       return makeNode(type, sizeof(SIndefRowsFuncPhysiNode));
     case QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC:
@@ -951,7 +953,8 @@ void nodesDestroyNode(SNode* pNode) {
       nodesDestroyNode(pPhyNode->pStateKey);
       break;
     }
-    case QUERY_NODE_PHYSICAL_PLAN_PARTITION: {
+    case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION: {
       SPartitionPhysiNode* pPhyNode = (SPartitionPhysiNode*)pNode;
       destroyPhysiNode((SPhysiNode*)pPhyNode);
       nodesDestroyList(pPhyNode->pExprs);

source/libs/parser/src/parAstCreater.c

@@ -795,6 +795,20 @@ SNode* createSetOperator(SAstCreateContext* pCxt, ESetOperatorType type, SNode*
   return (SNode*)setOp;
 }
 
+static void updateWalOptionsDefault(SDatabaseOptions* pOptions) {
+  if (!pOptions->walRetentionPeriodIsSet) {
+    pOptions->walRetentionPeriod =
+        pOptions->replica > 1 ? TSDB_REPS_DEF_DB_WAL_RET_PERIOD : TSDB_REP_DEF_DB_WAL_RET_PERIOD;
+  }
+  if (!pOptions->walRetentionSizeIsSet) {
+    pOptions->walRetentionSize = pOptions->replica > 1 ? TSDB_REPS_DEF_DB_WAL_RET_SIZE : TSDB_REP_DEF_DB_WAL_RET_SIZE;
+  }
+  if (!pOptions->walRollPeriodIsSet) {
+    pOptions->walRollPeriod =
+        pOptions->replica > 1 ? TSDB_REPS_DEF_DB_WAL_ROLL_PERIOD : TSDB_REP_DEF_DB_WAL_ROLL_PERIOD;
+  }
+}
+
 SNode* createDefaultDatabaseOptions(SAstCreateContext* pCxt) {
   CHECK_PARSER_STATUS(pCxt);
   SDatabaseOptions* pOptions = (SDatabaseOptions*)nodesMakeNode(QUERY_NODE_DATABASE_OPTIONS);
@@ -819,9 +833,7 @@ SNode* createDefaultDatabaseOptions(SAstCreateContext* pCxt) {
   pOptions->numOfVgroups = TSDB_DEFAULT_VN_PER_DB;
   pOptions->singleStable = TSDB_DEFAULT_DB_SINGLE_STABLE;
   pOptions->schemaless = TSDB_DEFAULT_DB_SCHEMALESS;
-  pOptions->walRetentionPeriod = TSDB_DEFAULT_DB_WAL_RETENTION_PERIOD;
-  pOptions->walRetentionSize = TSDB_DEFAULT_DB_WAL_RETENTION_SIZE;
-  pOptions->walRollPeriod = TSDB_DEFAULT_DB_WAL_ROLL_PERIOD;
+  updateWalOptionsDefault(pOptions);
   pOptions->walSegmentSize = TSDB_DEFAULT_DB_WAL_SEGMENT_SIZE;
   return (SNode*)pOptions;
 }
@@ -859,78 +871,83 @@ SNode* createAlterDatabaseOptions(SAstCreateContext* pCxt) {
 
 SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOptionType type, void* pVal) {
   CHECK_PARSER_STATUS(pCxt);
+  SDatabaseOptions* pDbOptions = (SDatabaseOptions*)pOptions;
   switch (type) {
     case DB_OPTION_BUFFER:
-      ((SDatabaseOptions*)pOptions)->buffer = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->buffer = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_CACHEMODEL:
-      COPY_STRING_FORM_STR_TOKEN(((SDatabaseOptions*)pOptions)->cacheModelStr, (SToken*)pVal);
+      COPY_STRING_FORM_STR_TOKEN(pDbOptions->cacheModelStr, (SToken*)pVal);
       break;
     case DB_OPTION_CACHESIZE:
-      ((SDatabaseOptions*)pOptions)->cacheLastSize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->cacheLastSize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_COMP:
-      ((SDatabaseOptions*)pOptions)->compressionLevel = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->compressionLevel = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_DAYS: {
       SToken* pToken = pVal;
       if (TK_NK_INTEGER == pToken->type) {
-        ((SDatabaseOptions*)pOptions)->daysPerFile = taosStr2Int32(pToken->z, NULL, 10) * 1440;
+        pDbOptions->daysPerFile = taosStr2Int32(pToken->z, NULL, 10) * 1440;
       } else {
-        ((SDatabaseOptions*)pOptions)->pDaysPerFile = (SValueNode*)createDurationValueNode(pCxt, pToken);
+        pDbOptions->pDaysPerFile = (SValueNode*)createDurationValueNode(pCxt, pToken);
       }
       break;
     }
     case DB_OPTION_FSYNC:
-      ((SDatabaseOptions*)pOptions)->fsyncPeriod = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->fsyncPeriod = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_MAXROWS:
-      ((SDatabaseOptions*)pOptions)->maxRowsPerBlock = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->maxRowsPerBlock = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_MINROWS:
-      ((SDatabaseOptions*)pOptions)->minRowsPerBlock = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->minRowsPerBlock = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_KEEP:
-      ((SDatabaseOptions*)pOptions)->pKeep = pVal;
+      pDbOptions->pKeep = pVal;
       break;
     case DB_OPTION_PAGES:
-      ((SDatabaseOptions*)pOptions)->pages = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->pages = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_PAGESIZE:
-      ((SDatabaseOptions*)pOptions)->pagesize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->pagesize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_PRECISION:
-      COPY_STRING_FORM_STR_TOKEN(((SDatabaseOptions*)pOptions)->precisionStr, (SToken*)pVal);
+      COPY_STRING_FORM_STR_TOKEN(pDbOptions->precisionStr, (SToken*)pVal);
       break;
     case DB_OPTION_REPLICA:
-      ((SDatabaseOptions*)pOptions)->replica = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->replica = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
+      updateWalOptionsDefault(pDbOptions);
       break;
     case DB_OPTION_STRICT:
-      COPY_STRING_FORM_STR_TOKEN(((SDatabaseOptions*)pOptions)->strictStr, (SToken*)pVal);
+      COPY_STRING_FORM_STR_TOKEN(pDbOptions->strictStr, (SToken*)pVal);
       break;
     case DB_OPTION_WAL:
-      ((SDatabaseOptions*)pOptions)->walLevel = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walLevel = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_VGROUPS:
-      ((SDatabaseOptions*)pOptions)->numOfVgroups = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->numOfVgroups = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_SINGLE_STABLE:
-      ((SDatabaseOptions*)pOptions)->singleStable = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->singleStable = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
       break;
     case DB_OPTION_RETENTIONS:
-      ((SDatabaseOptions*)pOptions)->pRetentions = pVal;
+      pDbOptions->pRetentions = pVal;
       break;
     case DB_OPTION_WAL_RETENTION_PERIOD:
-      ((SDatabaseOptions*)pOptions)->walRetentionPeriod = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walRetentionPeriod = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walRetentionPeriodIsSet = true;
       break;
     case DB_OPTION_WAL_RETENTION_SIZE:
-      ((SDatabaseOptions*)pOptions)->walRetentionSize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walRetentionSize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walRetentionSizeIsSet = true;
       break;
     case DB_OPTION_WAL_ROLL_PERIOD:
-      ((SDatabaseOptions*)pOptions)->walRollPeriod = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walRollPeriod = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walRollPeriodIsSet = true;
       break;
     case DB_OPTION_WAL_SEGMENT_SIZE:
-      ((SDatabaseOptions*)pOptions)->walSegmentSize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
+      pDbOptions->walSegmentSize = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
       break;
     default:
       break;
@@ -1251,7 +1268,8 @@ SNode* createUseDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName) {
 
 static bool needDbShowStmt(ENodeType type) {
   return QUERY_NODE_SHOW_TABLES_STMT == type || QUERY_NODE_SHOW_STABLES_STMT == type ||
-         QUERY_NODE_SHOW_VGROUPS_STMT == type;
+         QUERY_NODE_SHOW_VGROUPS_STMT == type || QUERY_NODE_SHOW_INDEXES_STMT == type ||
+         QUERY_NODE_SHOW_TAGS_STMT == type;
 }
 
 SNode* createShowStmt(SAstCreateContext* pCxt, ENodeType type) {
@@ -1264,7 +1282,7 @@ SNode* createShowStmt(SAstCreateContext* pCxt, ENodeType type) {
 SNode* createShowStmtWithCond(SAstCreateContext* pCxt, ENodeType type, SNode* pDbName, SNode* pTbName,
                               EOperatorType tableCondType) {
   CHECK_PARSER_STATUS(pCxt);
-  if (needDbShowStmt(type) && NULL == pDbName && NULL == pCxt->pQueryCxt->db) {
+  if (needDbShowStmt(type) && NULL == pDbName) {
     snprintf(pCxt->pQueryCxt->pMsg, pCxt->pQueryCxt->msgLen, "db not specified");
     pCxt->errCode = TSDB_CODE_PAR_SYNTAX_ERROR;
     return NULL;

source/libs/parser/src/parAstParser.c

@@ -374,8 +374,13 @@ static int32_t collectMetaKeyFromShowIndexes(SCollectMetaKeyCxt* pCxt, SShowStmt
 }
 
 static int32_t collectMetaKeyFromShowStables(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
-  return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_STABLES,
+  int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_STABLES,
                                          pCxt->pMetaCache);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = reserveUserAuthInCache(pCxt->pParseCxt->acctId, pCxt->pParseCxt->pUser,
+                                  ((SValueNode*)pStmt->pDbName)->literal, AUTH_TYPE_READ, pCxt->pMetaCache);
+  }
+  return code;
 }
 
 static int32_t collectMetaKeyFromShowStreams(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
@@ -387,11 +392,11 @@ static int32_t collectMetaKeyFromShowTables(SCollectMetaKeyCxt* pCxt, SShowStmt*
   int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_TABLES,
                                          pCxt->pMetaCache);
   if (TSDB_CODE_SUCCESS == code) {
-    if (NULL != pStmt->pDbName) {
     code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, ((SValueNode*)pStmt->pDbName)->literal, pCxt->pMetaCache);
-    } else {
-      code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, pCxt->pMetaCache);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = reserveUserAuthInCache(pCxt->pParseCxt->acctId, pCxt->pParseCxt->pUser,
+                                  ((SValueNode*)pStmt->pDbName)->literal, AUTH_TYPE_READ, pCxt->pMetaCache);
   }
   return code;
 }
@@ -400,11 +405,7 @@ static int32_t collectMetaKeyFromShowTags(SCollectMetaKeyCxt* pCxt, SShowStmt* p
   int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_TAGS,
                                          pCxt->pMetaCache);
   if (TSDB_CODE_SUCCESS == code) {
-    if (NULL != pStmt->pDbName) {
     code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, ((SValueNode*)pStmt->pDbName)->literal, pCxt->pMetaCache);
-    } else {
-      code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, pCxt->pMetaCache);
-    }
   }
   return code;
 }

source/libs/parser/src/parAuthenticator.c

@@ -96,6 +96,10 @@ static int32_t authInsert(SAuthCxt* pCxt, SInsertStmt* pInsert) {
   return code;
 }
 
+static int32_t authShowTables(SAuthCxt* pCxt, SShowStmt* pStmt) {
+  return checkAuth(pCxt, ((SValueNode*)pStmt->pDbName)->literal, AUTH_TYPE_READ);
+}
+
 static int32_t authShowCreateTable(SAuthCxt* pCxt, SShowCreateTableStmt* pStmt) {
   return checkAuth(pCxt, pStmt->dbName, AUTH_TYPE_READ);
 }
@@ -127,6 +131,9 @@ static int32_t authQuery(SAuthCxt* pCxt, SNode* pStmt) {
     case QUERY_NODE_SHOW_VNODES_STMT:
     case QUERY_NODE_SHOW_SCORES_STMT:
       return !pCxt->pParseCxt->enableSysInfo ? TSDB_CODE_PAR_PERMISSION_DENIED : TSDB_CODE_SUCCESS;
+    case QUERY_NODE_SHOW_TABLES_STMT:
+    case QUERY_NODE_SHOW_STABLES_STMT:
+      return authShowTables(pCxt, (SShowStmt*)pStmt);
     case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
     case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
       return authShowCreateTable(pCxt, (SShowCreateTableStmt*)pStmt);

source/libs/parser/src/parInsert.c

@@ -786,6 +786,7 @@ static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTa
   if (sname) pTbReq->ctb.name = strdup(sname);
   pTbReq->ctb.pTag = (uint8_t*)pTag;
   pTbReq->ctb.tagName = taosArrayDup(tagName);
+  pTbReq->ttl = TSDB_DEFAULT_TABLE_TTL;
   pTbReq->commentLen = -1;
 
   return;
@@ -1121,6 +1122,40 @@ static int32_t ignoreAutoCreateTableClause(SInsertParseContext* pCxt) {
   return code;
 }
 
+static int32_t parseTableOptions(SInsertParseContext* pCxt) {
+  do {
+    int32_t index = 0;
+    SToken  sToken;
+    NEXT_TOKEN_KEEP_SQL(pCxt->pSql, sToken, index);
+    if (TK_TTL == sToken.type) {
+      pCxt->pSql += index;
+      NEXT_TOKEN(pCxt->pSql, sToken);
+      if (TK_NK_INTEGER != sToken.type) {
+        return buildSyntaxErrMsg(&pCxt->msg, "Invalid option ttl", sToken.z);
+      }
+      pCxt->createTblReq.ttl = taosStr2Int32(sToken.z, NULL, 10);
+    } else if (TK_COMMENT == sToken.type) {
+      pCxt->pSql += index;
+      NEXT_TOKEN(pCxt->pSql, sToken);
+      if (TK_NK_STRING != sToken.type) {
+        return buildSyntaxErrMsg(&pCxt->msg, "Invalid option comment", sToken.z);
+      }
+      if (sToken.n >= TSDB_TB_COMMENT_LEN) {
+        return buildSyntaxErrMsg(&pCxt->msg, "comment too long", sToken.z);
+      }
+      int32_t len = trimString(sToken.z, sToken.n, pCxt->tmpTokenBuf, TSDB_TB_COMMENT_LEN);
+      pCxt->createTblReq.comment = strndup(pCxt->tmpTokenBuf, len);
+      if (NULL == pCxt->createTblReq.comment) {
+        return TSDB_CODE_OUT_OF_MEMORY;
+      }
+      pCxt->createTblReq.commentLen = len;
+    } else {
+      break;
+    }
+  } while (1);
+  return TSDB_CODE_SUCCESS;
+}
+
 // pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
 static int32_t parseUsingClause(SInsertParseContext* pCxt, int32_t tbNo, SName* name, char* tbFName) {
   int32_t      len = strlen(tbFName);
@@ -1172,7 +1207,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, int32_t tbNo, SName*
     return buildSyntaxErrMsg(&pCxt->msg, ") is expected", sToken.z);
   }
 
-  return TSDB_CODE_SUCCESS;
+  return parseTableOptions(pCxt);
 }
 
 static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks, int16_t timePrec, bool* gotRow,
@@ -1700,6 +1735,21 @@ static int32_t skipValuesClause(SInsertParseSyntaxCxt* pCxt) {
 
 static int32_t skipTagsClause(SInsertParseSyntaxCxt* pCxt) { return skipParentheses(pCxt); }
 
+static int32_t skipTableOptions(SInsertParseSyntaxCxt* pCxt) {
+  do {
+    int32_t index = 0;
+    SToken  sToken;
+    NEXT_TOKEN_KEEP_SQL(pCxt->pSql, sToken, index);
+    if (TK_TTL == sToken.type || TK_COMMENT == sToken.type) {
+      pCxt->pSql += index;
+      NEXT_TOKEN(pCxt->pSql, sToken);
+    } else {
+      break;
+    }
+  } while (1);
+  return TSDB_CODE_SUCCESS;
+}
+
 // pSql -> [(tag1_name, ...)] TAGS (tag1_value, ...)
 static int32_t skipUsingClause(SInsertParseSyntaxCxt* pCxt) {
   SToken sToken;
@@ -1718,6 +1768,7 @@ static int32_t skipUsingClause(SInsertParseSyntaxCxt* pCxt) {
     return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
   }
   CHECK_CODE(skipTagsClause(pCxt));
+  CHECK_CODE(skipTableOptions(pCxt));
 
   return TSDB_CODE_SUCCESS;
 }

source/libs/parser/test/parInitialCTest.cpp

@@ -111,9 +111,9 @@ TEST_F(ParserInitialCTest, createDatabase) {
     expect.numOfVgroups = TSDB_DEFAULT_VN_PER_DB;
     expect.numOfStables = TSDB_DEFAULT_DB_SINGLE_STABLE;
     expect.schemaless = TSDB_DEFAULT_DB_SCHEMALESS;
-    expect.walRetentionPeriod = TSDB_DEFAULT_DB_WAL_RETENTION_PERIOD;
-    expect.walRetentionSize = TSDB_DEFAULT_DB_WAL_RETENTION_SIZE;
-    expect.walRollPeriod = TSDB_DEFAULT_DB_WAL_ROLL_PERIOD;
+    expect.walRetentionPeriod = TSDB_REP_DEF_DB_WAL_RET_PERIOD;
+    expect.walRetentionSize = TSDB_REP_DEF_DB_WAL_RET_SIZE;
+    expect.walRollPeriod = TSDB_REP_DEF_DB_WAL_ROLL_PERIOD;
     expect.walSegmentSize = TSDB_DEFAULT_DB_WAL_SEGMENT_SIZE;
   };
 
@@ -266,6 +266,14 @@ TEST_F(ParserInitialCTest, createDatabase) {
       "DURATION 100m "
       "KEEP 1440m,300h,400d ");
   clearCreateDbReq();
+
+  setCreateDbReqFunc("wxy_db", 1);
+  setDbReplicaFunc(3);
+  setDbWalRetentionPeriod(TSDB_REPS_DEF_DB_WAL_RET_PERIOD);
+  setDbWalRetentionSize(TSDB_REPS_DEF_DB_WAL_RET_SIZE);
+  setDbWalRollPeriod(TSDB_REPS_DEF_DB_WAL_ROLL_PERIOD);
+  run("CREATE DATABASE IF NOT EXISTS wxy_db REPLICA 3");
+  clearCreateDbReq();
 }
 
 TEST_F(ParserInitialCTest, createDatabaseSemanticCheck) {

source/libs/planner/src/planOptimizer.c

@@ -16,6 +16,7 @@
 #include "filter.h"
 #include "functionMgt.h"
 #include "planInt.h"
+#include "tglobal.h"
 #include "ttime.h"
 
 #define OPTIMIZE_FLAG_MASK(n) (1 << n)
@@ -2410,7 +2411,7 @@ static const SOptimizeRule optimizeRuleSet[] = {
 static const int32_t optimizeRuleNum = (sizeof(optimizeRuleSet) / sizeof(SOptimizeRule));
 
 static void dumpLogicSubplan(const char* pRuleName, SLogicSubplan* pSubplan) {
-  if (0 == (qDebugFlag & DEBUG_DEBUG)) {
+  if (!tsQueryPlannerTrace) {
     return;
   }
   char* pStr = NULL;

source/libs/planner/src/planPhysiCreater.c

@@ -1324,7 +1324,8 @@ static int32_t createSortPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
 static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren,
                                         SPartitionLogicNode* pPartLogicNode, SPhysiNode** pPhyNode) {
   SPartitionPhysiNode* pPart =
-      (SPartitionPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pPartLogicNode, QUERY_NODE_PHYSICAL_PLAN_PARTITION);
+      (SPartitionPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pPartLogicNode,
+          pCxt->pPlanCxt->streamQuery ? QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION : QUERY_NODE_PHYSICAL_PLAN_PARTITION);
   if (NULL == pPart) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }

source/libs/planner/src/planSpliter.c

@@ -1427,7 +1427,7 @@ static const SSplitRule splitRuleSet[] = {
 static const int32_t splitRuleNum = (sizeof(splitRuleSet) / sizeof(SSplitRule));
 
 static void dumpLogicSubplan(const char* pRuleName, SLogicSubplan* pSubplan) {
-  if (0 == (qDebugFlag & DEBUG_DEBUG)) {
+  if (!tsQueryPlannerTrace) {
     return;
   }
   char* pStr = NULL;

source/libs/planner/src/planner.c

@@ -17,9 +17,10 @@
 
 #include "planInt.h"
 #include "scalar.h"
+#include "tglobal.h"
 
 static void dumpQueryPlan(SQueryPlan* pPlan) {
-  if (0 == (qDebugFlag & DEBUG_DEBUG)) {
+  if (!tsQueryPlannerTrace) {
     return;
   }
   char* pStr = NULL;

source/libs/stream/src/streamUpdate.c

@@ -170,8 +170,17 @@ bool updateInfoIsUpdated(SUpdateInfo *pInfo, uint64_t tableId, TSKEY ts) {
   if (ts < maxTs - pInfo->watermark) {
     // this window has been closed.
     if (pInfo->pCloseWinSBF) {
-      return tScalableBfPut(pInfo->pCloseWinSBF, &ts, sizeof(TSKEY));
+      res = tScalableBfPut(pInfo->pCloseWinSBF, &ts, sizeof(TSKEY));
+      if (res == TSDB_CODE_SUCCESS) {
+        return false;
+      } else {
+         qDebug("===stream===Update close window sbf. tableId:%" PRIu64 ", maxTs:%" PRIu64 ", mapMaxTs:%" PRIu64 ", ts:%" PRIu64, tableId,
+                maxTs, *pMapMaxTs, ts);
+        return true;
       }
+    }
+    qDebug("===stream===Update close window. tableId:%" PRIu64 ", maxTs:%" PRIu64 ", mapMaxTs:%" PRIu64 ", ts:%" PRIu64, tableId,
+           maxTs, *pMapMaxTs, ts);
     return true;
   }
 
@@ -193,7 +202,7 @@ bool updateInfoIsUpdated(SUpdateInfo *pInfo, uint64_t tableId, TSKEY ts) {
   }
 
   if (ts < pInfo->minTS) {
-    qDebug("===stream===Update. tableId:%" PRIu64 ", maxTs:%" PRIu64 ", mapMaxTs:%" PRIu64 ", ts:%" PRIu64, tableId,
+    qDebug("===stream===Update min ts. tableId:%" PRIu64 ", maxTs:%" PRIu64 ", mapMaxTs:%" PRIu64 ", ts:%" PRIu64, tableId,
            maxTs, *pMapMaxTs, ts);
     return true;
   } else if (res == TSDB_CODE_SUCCESS) {

source/os/src/osDir.c

@@ -160,6 +160,66 @@ int32_t taosMulMkDir(const char *dirname) {
   return code;
 }
 
+int32_t taosMulModeMkDir(const char *dirname, int mode) {
+  if (dirname == NULL) return -1;
+  char    temp[1024];
+  char   *pos = temp;
+  int32_t code = 0;
+#ifdef WINDOWS
+  taosRealPath(dirname, temp, sizeof(temp));
+  if (temp[1] == ':') pos += 3;
+#else
+  strcpy(temp, dirname);
+#endif
+
+  if (taosDirExist(temp)) {
+    chmod(temp, mode);
+    return code;
+  }
+
+  if (strncmp(temp, TD_DIRSEP, 1) == 0) {
+    pos += 1;
+  } else if (strncmp(temp, "." TD_DIRSEP, 2) == 0) {
+    pos += 2;
+  }
+
+  for (; *pos != '\0'; pos++) {
+    if (*pos == TD_DIRSEP[0]) {
+      *pos = '\0';
+#ifdef WINDOWS
+      code = _mkdir(temp, mode);
+#else
+      code = mkdir(temp, mode);
+#endif
+      if (code < 0 && errno != EEXIST) {
+        terrno = TAOS_SYSTEM_ERROR(errno);
+        return code;
+      }
+      *pos = TD_DIRSEP[0];
+    }
+  }
+
+  if (*(pos - 1) != TD_DIRSEP[0]) {
+#ifdef WINDOWS
+    code = _mkdir(temp, mode);
+#else
+    code = mkdir(temp, mode);
+#endif
+    if (code < 0 && errno != EEXIST) {
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      return code;
+    }
+  }
+
+  if (code < 0 && errno == EEXIST) {
+    chmod(temp, mode);
+    return 0;
+  }
+
+  chmod(temp, mode);
+  return code;
+}
+
 void taosRemoveOldFiles(const char *dirname, int32_t keepDays) {
   TdDirPtr pDir = taosOpenDir(dirname);
   if (pDir == NULL) return;

tests/script/tsim/stream/basic1.sim

@@ -462,10 +462,10 @@ if $data25 != 3 then
   return -1
 endi
 
-sql create database test2 vgroups 1
-sql select * from information_schema.ins_databases
+sql create database test2 vgroups 1;
+sql select * from information_schema.ins_databases;
 
-sql use test2
+sql use test2;
 sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
 sql create table t1 using st tags(1,1,1);
 sql create table t2 using st tags(2,2,2);

tests/script/tsim/stream/partitionbyColumn0.sim

+$loop_all = 0
+looptest:
+
+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+sql drop stream if exists streams0;
+sql drop stream if exists streams1;
+sql drop stream if exists streams2;
+sql drop stream if exists streams3;
+sql drop stream if exists streams4;
+sql drop database if exists test;
+sql create database test  vgroups 1;
+sql use test;
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+sql create stream streams0 trigger at_once into streamt as select  _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a interval(10s);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+
+$loop_count = 0
+
+loop0:
+sleep 100
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != NULL then
+  print =====data02=$data02
+  goto loop0
+endi
+
+
+sql insert into t1 values(1648791213000,1,2,3,1.0);
+
+$loop_count = 0
+
+loop1:
+sleep 100
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop1
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+
+$loop_count = 0
+
+loop2:
+sleep 100
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop2
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t1 values(1648791213001,2,2,3,1.0);
+sql insert into t1 values(1648791213002,2,2,3,1.0);
+sql insert into t1 values(1648791213002,1,2,3,1.0);
+
+$loop_count = 0
+
+loop3:
+sleep 100
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop3
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop3
+endi
+
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop3
+endi
+
+if $data12 != 2 then
+  print =====data12=$data12
+  goto loop3
+endi
+
+sql insert into t1 values(1648791223000,1,2,3,1.0);
+sql insert into t1 values(1648791223001,1,2,3,1.0);
+sql insert into t1 values(1648791223002,3,2,3,1.0);
+sql insert into t1 values(1648791223003,3,2,3,1.0);
+sql insert into t1 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
+
+$loop_count = 0
+
+loop4:
+sleep 100
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop4
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop4
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop4
+endi
+
+if $data12 != 2 then
+  print =====data12=$data12
+  goto loop4
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop4
+endi
+
+if $data22 != 1 then
+  print =====data22=$data22
+  goto loop4
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop4
+endi
+
+if $data32 != 2 then
+  print =====data32=$data32
+  goto loop4
+endi
+
+if $data41 != 1 then
+  print =====data41=$data41
+  goto loop4
+endi
+
+if $data42 != 3 then
+  print =====data42=$data42
+  goto loop4
+endi
+
+sql drop stream if exists streams1;
+sql drop database if exists test1;
+sql create database test1  vgroups 1;
+sql use test1;
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+sql create stream streams1 trigger at_once into streamt1 as select  _wstart c1, count(*) c2, max(c) c3, _group_key(a+b) c4 from t1 partition by a+b interval(10s);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,1,2,1,1.0);
+sql insert into t1 values(1648791213001,2,1,2,2.0);
+sql insert into t1 values(1648791213001,1,2,3,2.0);
+
+$loop_count = 0
+
+loop5:
+sleep 100
+sql select * from streamt1 order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop5
+endi
+
+sql insert into t1 values(1648791223000,1,2,4,2.0);
+sql insert into t1 values(1648791223001,1,2,5,2.0);
+sql insert into t1 values(1648791223002,1,2,5,2.0);
+sql insert into t1 values(1648791213001,1,1,6,2.0) (1648791223002,1,1,7,2.0);
+
+$loop_count = 0
+
+loop6:
+sleep 100
+sql select * from streamt1 order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop6
+endi
+
+if $data02 != 6 then
+  print =====data02=$data02
+  goto loop6
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop6
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop6
+endi
+
+if $data21 != 1 then
+  print =====data21=$data21
+  goto loop6
+endi
+
+if $data22 != 7 then
+  print =====data22=$data22
+  goto loop6
+endi
+
+if $data31 != 2 then
+  print =====data31=$data31
+  goto loop6
+endi
+
+if $data32 != 5 then
+  print =====data32=$data32
+  goto loop6
+endi
+
+sql drop stream if exists streams2;
+sql drop database if exists test2;
+sql create database test2  vgroups 4;
+sql use test2;
+sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
+sql create table t1 using st tags(1,1,1);
+sql create table t2 using st tags(2,2,2);
+sql create stream streams2 trigger at_once into test.streamt2 as select  _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL);
+
+$loop_count = 0
+
+loop7:
+sleep 100
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop7
+endi
+
+if $data02 != NULL then
+  print =====data02=$data02
+  goto loop7
+endi
+
+sql insert into t1 values(1648791213000,1,2,3,1.0);
+sql insert into t2 values(1648791213000,1,2,3,1.0);
+
+$loop_count = 0
+
+loop8:
+sleep 100
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop8
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop8
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t2 values(1648791213000,2,2,3,1.0);
+
+$loop_count = 0
+
+loop9:
+sleep 100
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop9
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop9
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t1 values(1648791213001,2,2,3,1.0);
+sql insert into t1 values(1648791213002,2,2,3,1.0);
+sql insert into t1 values(1648791213002,1,2,3,1.0);
+sql insert into t2 values(1648791213000,2,2,3,1.0);
+sql insert into t2 values(1648791213001,2,2,3,1.0);
+sql insert into t2 values(1648791213002,2,2,3,1.0);
+sql insert into t2 values(1648791213002,1,2,3,1.0);
+
+$loop_count = 0
+
+loop10:
+sleep 100
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop10
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop10
+endi
+
+if $data11 != 4 thenloop4
+  print =====data11=$data11
+  goto loop10
+endi
+
+if $data12 != 2 then
+  print =====data12=$data12
+  goto loop10
+endi
+
+sql insert into t1 values(1648791223000,1,2,3,1.0);
+sql insert into t1 values(1648791223001,1,2,3,1.0);
+sql insert into t1 values(1648791223002,3,2,3,1.0);
+sql insert into t1 values(1648791223003,3,2,3,1.0);
+sql insert into t1 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
+sql insert into t2 values(1648791223000,1,2,3,1.0);
+sql insert into t2 values(1648791223001,1,2,3,1.0);
+sql insert into t2 values(1648791223002,3,2,3,1.0);
+sql insert into t2 values(1648791223003,3,2,3,1.0);
+sql insert into t2 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
+
+$loop_count = 0
+
+loop11:
+sleep 100
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop11
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop11
+endi
+
+if $data11 != 4 then
+  print =====data11=$data11
+  goto loop11
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop11
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop11
+endi
+
+if $data22 != 2 then
+  print =====data22=$data22
+  goto loop11
+endi
+
+if $data31 != 2 then
+  print =====data31=$data31
+  goto loop11
+endi
+
+if $data32 != 3 then
+  print =====data32=$data32
+  goto loop11
+endi
+
+if $data41 != 4 then
+  print =====data41=$data41
+  goto loop11
+endi
+
+if $data42 != 1 then
+  print =====data42=$data42
+  goto loop11
+endi
+
+sql drop stream if exists streams4;
+sql drop database if exists test4;
+sql create database test4  vgroups 4;
+sql use test4;
+sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
+sql create table t1 using st tags(1,1,1);
+sql create table t2 using st tags(2,2,2);
+sql create table t3 using st tags(2,2,2);
+sql create table t4 using st tags(2,2,2);
+sql create stream streams4 trigger at_once into test.streamt4 as select  _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s);
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t2 values(1648791213000,2,2,3,1.0);
+sql insert into t3 values(1648791213000,2,2,3,1.0);
+sql insert into t4 values(1648791213000,2,2,3,1.0);
+sql insert into t4 values(1648791213000,1,2,3,1.0);
+
+$loop_count = 0
+
+loop13:
+sleep 100
+sql select * from test.streamt4 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $rows != 2 then
+  print =====rows=$rows
+  goto loop13
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop13
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop13
+endi
+
+if $data11 != 3 then
+  print =====data11=$data11
+  goto loop13
+endi
+
+if $data12 != 2 then
+  print =====data12=$data12
+  goto loop13
+endi
+
+sql insert into t4 values(1648791213000,2,2,3,1.0);
+sql insert into t1 values(1648791233000,2,2,3,1.0);
+
+
+sql insert into t1 values(1648791213000,1,2,3,1.0); 
+
+$loop_count = 0
+
+loop14:
+sleep 100
+sql select * from test.streamt4 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
+if $rows != 3 then
+  print =====rows=$rows
+  goto loop14
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop14
+endi
+
+if $data11 != 3 then
+  print =====data11=$data11
+  goto loop14
+endi
+
+if $data21 != 1 then
+  print =====data21=$data21
+  goto loop14
+endi
+
+$loop_all = $loop_all + 1
+print ============loop_all=$loop_all
+
+system sh/stop_dnodes.sh
+
+#goto looptest
\ No newline at end of file

tests/script/tsim/stream/partitionbyColumn1.sim

+$loop_all = 0
+looptest:
+
+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+sql drop stream if exists streams0;
+sql drop stream if exists streams1;
+sql drop stream if exists streams2;
+sql drop stream if exists streams3;
+sql drop stream if exists streams4;
+sql drop database if exists test;
+sql create database test  vgroups 1;
+sql use test;
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+sql create stream streams0 trigger at_once into streamt as select  _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a session(ts, 5s);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+
+$loop_count = 0
+
+loop0:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != NULL then
+  print =====data02=$data02
+  goto loop0
+endi
+
+
+sql insert into t1 values(1648791213000,1,2,3,1.0);
+
+loop1:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop1
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+
+loop2:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop2
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t1 values(1648791213001,2,2,3,1.0);
+sql insert into t1 values(1648791213002,2,2,3,1.0);
+sql insert into t1 values(1648791213002,1,2,3,1.0);
+
+loop3:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop3
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop3
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop3
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop3
+endi
+
+sql insert into t1 values(1648791223000,1,2,3,1.0);
+sql insert into t1 values(1648791223001,1,2,3,1.0);
+sql insert into t1 values(1648791223002,3,2,3,1.0);
+sql insert into t1 values(1648791223003,3,2,3,1.0);
+sql insert into t1 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
+
+loop4:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop4
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop4
+endi
+
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop4
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop4
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop4
+endi
+
+if $data22 != 1 then
+  print =====data22=$data22
+  goto loop4
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop4
+endi
+
+if $data32 != 2 then
+  print =====data32=$data32
+  goto loop4
+endi
+
+if $data41 != 1 then
+  print =====data41=$data41
+  goto loop4
+endi
+
+if $data42 != 3 then
+  print =====data42=$data42
+  goto loop4
+endi
+
+sql drop database if exists test1;
+sql create database test1  vgroups 1;
+sql use test1;
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+sql create stream streams1 trigger at_once into streamt1 as select  _wstart c1, count(*) c2, max(c) c3, _group_key(a+b) c4 from t1 partition by a+b session(ts, 5s);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,1,2,1,1.0);
+sql insert into t1 values(1648791213001,2,1,2,2.0);
+sql insert into t1 values(1648791213001,1,2,3,2.0);
+
+$loop_count = 0
+
+loop5:
+sleep 300
+sql select * from streamt1 order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop5
+endi
+
+sql insert into t1 values(1648791223000,1,2,4,2.0);
+sql insert into t1 values(1648791223001,1,2,5,2.0);
+sql insert into t1 values(1648791223002,1,2,5,2.0);
+sql insert into t1 values(1648791213001,1,1,6,2.0) (1648791223002,1,1,7,2.0);
+
+loop6:
+sleep 300
+sql select * from streamt1 order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop6
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop6
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop6
+endi
+
+if $data12 != 6 then
+  print =====data12=$data12
+  goto loop6
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop6
+endi
+
+if $data22 != 5 then
+  print =====data22=$data22
+  goto loop6
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop6
+endi
+
+if $data32 != 7 then
+  print =====data32=$data32
+  goto loop6
+endi
+
+sql drop database if exists test2;
+sql create database test2  vgroups 4;
+sql use test2;
+sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
+sql create table t1 using st tags(1,1,1);
+sql create table t2 using st tags(2,2,2);
+sql create stream streams2 trigger at_once into test.streamt2 as select  _wstart c1, count(*) c2, max(a) c3 from st partition by a session(ts, 5s);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL);
+
+$loop_count = 0
+
+loop7:
+sleep 300
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop7
+endi
+
+if $data02 != NULL then
+  print =====data02=$data02
+  goto loop7
+endi
+
+sql insert into t1 values(1648791213000,1,2,3,1.0);
+sql insert into t2 values(1648791213000,1,2,3,1.0);
+
+loop8:
+sleep 300
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop8
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop8
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t2 values(1648791213000,2,2,3,1.0);
+loop9:
+sleep 300
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop9
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop9
+endi
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t1 values(1648791213001,2,2,3,1.0);
+sql insert into t1 values(1648791213002,2,2,3,1.0);
+sql insert into t1 values(1648791213002,1,2,3,1.0);
+sql insert into t2 values(1648791213000,2,2,3,1.0);
+sql insert into t2 values(1648791213001,2,2,3,1.0);
+sql insert into t2 values(1648791213002,2,2,3,1.0);
+sql insert into t2 values(1648791213002,1,2,3,1.0);
+
+loop10:
+sleep 300
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 4 then
+  print =====data01=$data01
+  goto loop10
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop10
+endi
+
+if $data11 != 2 thenloop4
+  print =====data11=$data11
+  goto loop3
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop10
+endi
+
+sql insert into t1 values(1648791223000,1,2,3,1.0);
+sql insert into t1 values(1648791223001,1,2,3,1.0);
+sql insert into t1 values(1648791223002,3,2,3,1.0);
+sql insert into t1 values(1648791223003,3,2,3,1.0);
+sql insert into t1 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
+sql insert into t2 values(1648791223000,1,2,3,1.0);
+sql insert into t2 values(1648791223001,1,2,3,1.0);
+sql insert into t2 values(1648791223002,3,2,3,1.0);
+sql insert into t2 values(1648791223003,3,2,3,1.0);
+sql insert into t2 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
+
+loop11:
+sleep 300
+sql select * from test.streamt2 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop11
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop11
+endi
+
+if $data11 != 4 then
+  print =====data11=$data11
+  goto loop11
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop11
+endi
+
+if $data21 != 4 then
+  print =====data21=$data21
+  goto loop11
+endi
+
+if $data22 != 1 then
+  print =====data22=$data22
+  goto loop11
+endi
+
+if $data31 != 2 then
+  print =====data31=$data31
+  goto loop11
+endi
+
+if $data32 != 2 then
+  print =====data32=$data32
+  goto loop11
+endi
+
+if $data41 != 2 then
+  print =====data41=$data41
+  goto loop11
+endi
+
+if $data42 != 3 then
+  print =====data42=$data42
+  goto loop11
+endi
+
+sql drop database if exists test4;
+sql create database test4  vgroups 4;
+sql use test4;
+sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
+sql create table t1 using st tags(1,1,1);
+sql create table t2 using st tags(2,2,2);
+sql create table t3 using st tags(2,2,2);
+sql create table t4 using st tags(2,2,2);
+sql create stream streams4 trigger at_once into test.streamt4 as select  _wstart c1, count(*) c2, max(a) c3 from st partition by a session(ts, 5s);
+
+sql insert into t1 values(1648791213000,2,2,3,1.0);
+sql insert into t2 values(1648791213000,2,2,3,1.0);
+sql insert into t3 values(1648791213000,2,2,3,1.0);
+sql insert into t4 values(1648791213000,2,2,3,1.0);
+sql insert into t4 values(1648791213000,1,2,3,1.0);
+
+$loop_count = 0
+
+loop13:
+sleep 300
+sql select * from test.streamt4 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $rows != 2 then
+  print =====rows=$rows
+  goto loop14
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop13
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop13
+endi
+
+if $data11 != 3 then
+  print =====data11=$data11
+  goto loop11
+endi
+
+if $data12 != 2 then
+  print =====data12=$data12
+  goto loop11
+endi
+
+sql insert into t4 values(1648791213000,2,2,3,1.0);
+sql insert into t1 values(1648791233000,2,2,3,1.0);
+
+
+sql insert into t1 values(1648791213000,1,2,3,1.0); 
+
+loop14:
+sleep 300
+sql select * from test.streamt4 order by c1, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $rows != 3 then
+  print =====rows=$rows
+  goto loop14
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop14
+endi
+
+if $data11 != 3 then
+  print =====data11=$data11
+  goto loop14
+endi
+
+if $data21 != 1 then
+  print =====data21=$data21
+  goto loop14
+endi
+
+system sh/stop_dnodes.sh
+
+$loop_all = $loop_all + 1
+print ============loop_all=$loop_all
+
+#goto looptest
\ No newline at end of file

tests/script/tsim/stream/partitionbyColumn2.sim

+$loop_all = 0
+looptest:
+
+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+sql drop database if exists test;
+sql create database test  vgroups 1;
+sql use test;
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+sql create stream streams0 trigger at_once into streamt as select  _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a state_window(b);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+
+$loop_count = 0
+
+loop0:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != NULL then
+  print =====data02=$data02
+  goto loop0
+endi
+
+
+sql insert into t1 values(1648791213000,1,1,3,1.0);
+
+loop1:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop1
+endi
+
+sql insert into t1 values(1648791213000,2,1,3,1.0);
+
+loop2:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop2
+endi
+
+sql insert into t1 values(1648791213000,2,1,3,1.0);
+sql insert into t1 values(1648791213001,2,1,3,1.0);
+sql insert into t1 values(1648791213002,2,1,3,1.0);
+sql insert into t1 values(1648791213002,1,1,3,1.0);
+
+loop3:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop3
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop3
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop3
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop3
+endi
+
+sql insert into t1 values(1648791223000,1,2,3,1.0);
+sql insert into t1 values(1648791223001,1,2,3,1.0);
+sql insert into t1 values(1648791223002,3,2,3,1.0);
+sql insert into t1 values(1648791223003,3,2,3,1.0);
+sql insert into t1 values(1648791213001,1,1,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
+
+loop4:
+sleep 300
+sql select * from streamt order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop4
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop4
+endi
+
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop4
+endi
+
+if $data12 != 1 then
+  print =====data12=$data12
+  goto loop4
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop4
+endi
+
+if $data22 != 1 then
+  print =====data22=$data22
+  goto loop4
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop4
+endi
+
+if $data32 != 2 then
+  print =====data32=$data32
+  goto loop4
+endi
+
+if $data41 != 1 then
+  print =====data41=$data41
+  goto loop4
+endi
+
+if $data42 != 3 then
+  print =====data42=$data42
+  goto loop4
+endi
+
+sql drop database if exists test1;
+sql create database test1  vgroups 1;
+sql use test1;
+sql create table t1(ts timestamp, a int, b int , c int, d int);
+sql create stream streams1 trigger at_once into streamt1 as select  _wstart c1, count(*) c2, max(d) c3, _group_key(a+b) c4 from t1 partition by a+b state_window(c);
+
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
+sql insert into t1 values(1648791213000,1,2,1,1);
+sql insert into t1 values(1648791213001,2,1,1,2);
+sql insert into t1 values(1648791213001,1,2,1,3);
+
+$loop_count = 0
+
+loop5:
+sleep 300
+sql select * from streamt1 order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop5
+endi
+
+sql insert into t1 values(1648791223000,1,2,2,4);
+sql insert into t1 values(1648791223001,1,2,2,5);
+sql insert into t1 values(1648791223002,1,2,2,6);
+sql insert into t1 values(1648791213001,1,1,1,7) (1648791223002,1,1,2,8);
+
+loop6:
+sleep 300
+sql select * from streamt1 order by c1, c4, c2, c3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop6
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop6
+endi
+
+if $data11 != 1 then
+  print =====data11=$data11
+  goto loop6
+endi
+
+if $data12 != 7 then
+  print =====data12=$data12
+  goto loop6
+endi
+
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop6
+endi
+
+if $data22 != 5 then
+  print =====data22=$data22
+  goto loop6
+endi
+
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop6
+endi
+
+if $data32 != 8 then
+  print =====data32=$data32
+  goto loop6
+endi
+
+system sh/stop_dnodes.sh
+
+$loop_all = $loop_all + 1
+print ============loop_all=$loop_all
+
+#goto looptest

tests/system-test/0-others/user_control.py

@@ -282,12 +282,12 @@ class TDTestCase:
                 use.error(f"insert into {DBNAME}.{CTBNAME} (ts) values (now())")
             elif check_priv == PRIVILEGES_WRITE:
                 use.query(f"use {DBNAME}")
-                use.query(f"show {DBNAME}.tables")
+                use.error(f"show {DBNAME}.tables")
                 use.error(f"select * from {DBNAME}.{CTBNAME}")
                 use.query(f"insert into {DBNAME}.{CTBNAME} (ts) values (now())")
             elif check_priv is None:
                 use.error(f"use {DBNAME}")
-                # use.error(f"show {DBNAME}.tables")
+                use.error(f"show {DBNAME}.tables")
                 use.error(f"show tables")
                 use.error(f"select * from {DBNAME}.{CTBNAME}")
                 use.error(f"insert into {DBNAME}.{CTBNAME} (ts) values (now())")