Merge branch '3.0' into feature/TD-14481-3.0

docs/en/14-reference/02-rest-api/02-rest-api.mdx

@@ -21,7 +21,7 @@ The following example is in an Ubuntu environment and uses the `curl` tool to ve
 The following example lists all databases on the host h1.taosdata.com. To use it in your environment, replace `h1.taosdata.com` and `6041` (the default port) with the actual running TDengine service FQDN and port number.
 
 ```html
-curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'show databases;' h1.taosdata.com:6041/rest/sql
+curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "show databases;" h1.taosdata.com:6041/rest/sql
 ```
 
 The following return value results indicate that the verification passed.
@@ -106,13 +106,13 @@ The HTTP request's BODY is a complete SQL command, and the data table in the SQL
 Use `curl` to initiate an HTTP request with a custom authentication method, with the following syntax.
 
 ```bash
-curl -H 'Authorization: Basic <TOKEN>' -d '<SQL>' <ip>:<PORT>/rest/sql/[db_name]
+curl -L -H "Authorization: Basic <TOKEN>" -d "<SQL>" <ip>:<PORT>/rest/sql/[db_name]
 ```
 
 Or
 
 ```bash
-curl -u username:password -d '<SQL>' <ip>:<PORT>/rest/sql/[db_name]
+curl -L -u username:password -d "<SQL>" <ip>:<PORT>/rest/sql/[db_name]
 ```
 
 where `TOKEN` is the string after Base64 encoding of `{username}:{password}`, e.g. `root:taosdata` is encoded as `cm9vdDp0YW9zZGF0YQ==`.
@@ -192,7 +192,7 @@ Response body:
 - query all records from table d1001 of database demo
 
   ```bash
-  curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'select * from demo.d1001' 192.168.0.1:6041/rest/sql
+  curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "select * from demo.d1001" 192.168.0.1:6041/rest/sql
   ```
 
   Response body:
@@ -218,7 +218,7 @@ Response body:
 - Create database demo:
 
   ```bash
-  curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'create database demo' 192.168.0.1:6041/rest/sql
+  curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "create database demo" 192.168.0.1:6041/rest/sql
   ```
 
   Response body:
@@ -240,7 +240,7 @@ Response body:
 When the HTTP request URL uses `/rest/sqlt`, the returned result set's timestamp value will be in Unix timestamp format, for example:
 
 ```bash
-curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'select * from demo.d1001' 192.168.0.1:6041/rest/sqlt
+curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "select * from demo.d1001" 192.168.0.1:6041/rest/sqlt
 ```
 
 Response body:
@@ -268,7 +268,7 @@ Response body:
 When the HTTP request URL uses `/rest/sqlutc`, the timestamp of the returned result set will be expressed as a UTC format, for example:
 
 ```bash
-  curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'select * from demo.t1' 192.168.0.1:6041/rest/sqlutc
+  curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "select * from demo.t1" 192.168.0.1:6041/rest/sqlutc
 ```
 
 Response body:

docs/en/14-reference/12-directory.md

@@ -26,7 +26,6 @@ All executable files of TDengine are in the _/usr/local/taos/bin_ directory by d
 - _remove.sh_: script to uninstall TDengine, please execute it carefully, link to the **rmtaos** command in the /usr/bin directory. Will remove the TDengine installation directory `/usr/local/taos`, but will keep `/etc/taos`, `/var/lib/taos`, `/var/log/taos`
 - _taosadapter_: server-side executable that provides RESTful services and accepts writing requests from a variety of other softwares
 - _tarbitrator_: provides arbitration for two-node cluster deployments
-- _run_taosd_and_taosadapter.sh_: script to start both taosd and taosAdapter
 - _TDinsight.sh_: script to download TDinsight and install it
 - _set_core.sh_: script for setting up the system to generate core dump files for easy debugging
 - _taosd-dump-cfg.gdb_: script to facilitate debugging of taosd's gdb execution.

docs/en/27-train-faq/03-docker.md

@@ -109,7 +109,7 @@ taos>
 It's also able to access the REST interface provided by TDengine in container from the host.
 
 ```
-curl -u root:taosdata -d 'show databases' 127.0.0.1:6041/rest/sql
+curl -L -u root:taosdata -d "show databases" 127.0.0.1:6041/rest/sql
 ```
 
 Output is like below:
@@ -147,7 +147,7 @@ docker run -d --name tdengine-taosd -p 6030-6042:6030-6042 -p 6030-6042:6030-604
 - Verify the REST interface:
 
 ```bash
-curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'show databases;' 127.0.0.1:6041/rest/sql
+curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "show databases;" 127.0.0.1:6041/rest/sql
 ```
 
 Below is an example output:

docs/zh/14-reference/02-rest-api/02-rest-api.mdx

@@ -21,7 +21,7 @@ RESTful 接口不依赖于任何 TDengine 的库，因此客户端不需要安
 下面示例是列出所有的数据库，请把 h1.taosdata.com 和 6041（缺省值）替换为实际运行的 TDengine 服务 FQDN 和端口号：
 
 ```html
-curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'show databases;' h1.taosdata.com:6041/rest/sql
+curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "show databases;" h1.taosdata.com:6041/rest/sql
 ```
 
 返回值结果如下表示验证通过：
@@ -106,13 +106,13 @@ HTTP 请求的 BODY 里就是一个完整的 SQL 语句，SQL 语句中的数据
 使用 `curl` 通过自定义身份认证方式来发起一个 HTTP Request，语法如下：
 
 ```bash
-curl -H 'Authorization: Basic <TOKEN>' -d '<SQL>' <ip>:<PORT>/rest/sql/[db_name]
+curl -L -H "Authorization: Basic <TOKEN>" -d "<SQL>" <ip>:<PORT>/rest/sql/[db_name]
 ```
 
 或者
 
 ```bash
-curl -u username:password -d '<SQL>' <ip>:<PORT>/rest/sql/[db_name]
+curl -L -u username:password -d "<SQL>" <ip>:<PORT>/rest/sql/[db_name]
 ```
 
 其中，`TOKEN` 为 `{username}:{password}` 经过 Base64 编码之后的字符串，例如 `root:taosdata` 编码后为 `cm9vdDp0YW9zZGF0YQ==`
@@ -192,7 +192,7 @@ curl http://192.168.0.1:6041/rest/login/root/taosdata
 - 在 demo 库里查询表 d1001 的所有记录：
 
   ```bash
-  curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'select * from demo.d1001' 192.168.0.1:6041/rest/sql
+  curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "select * from demo.d1001" 192.168.0.1:6041/rest/sql
   ```
 
   返回值：
@@ -218,7 +218,7 @@ curl http://192.168.0.1:6041/rest/login/root/taosdata
 - 创建库 demo：
 
   ```bash
-  curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'create database demo' 192.168.0.1:6041/rest/sql
+  curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "create database demo" 192.168.0.1:6041/rest/sql
   ```
 
   返回值：
@@ -240,7 +240,7 @@ curl http://192.168.0.1:6041/rest/login/root/taosdata
 HTTP 请求 URL 采用 `/rest/sqlt` 时，返回结果集的时间戳将采用 Unix 时间戳格式表示，例如
 
 ```bash
-curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'select * from demo.d1001' 192.168.0.1:6041/rest/sqlt
+curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "select * from demo.d1001" 192.168.0.1:6041/rest/sqlt
 ```
 
 返回结果：
@@ -268,7 +268,7 @@ curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'select * from demo.d1001
 HTTP 请求 URL 采用 `/rest/sqlutc` 时，返回结果集的时间戳将采用 UTC 时间字符串表示，例如
 
 ```bash
-  curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'select * from demo.t1' 192.168.0.1:6041/rest/sqlutc
+  curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "select * from demo.t1" 192.168.0.1:6041/rest/sqlutc
 ```
 
 返回值：

docs/zh/14-reference/12-directory.md

@@ -26,7 +26,6 @@ TDengine 的所有可执行文件默认存放在 _/usr/local/taos/bin_ 目录下
 - _remove.sh_：卸载 TDengine 的脚本，请谨慎执行，链接到/usr/bin 目录下的**rmtaos**命令。会删除 TDengine 的安装目录/usr/local/taos，但会保留/etc/taos、/var/lib/taos、/var/log/taos
 - _taosadapter_: 提供 RESTful 服务和接受其他多种软件写入请求的服务端可执行文件
 - _tarbitrator_: 提供双节点集群部署的仲裁功能
-- _run_taosd_and_taosadapter.sh_：同时启动 taosd 和 taosAdapter 的脚本
 - _TDinsight.sh_：用于下载 TDinsight 并安装的脚本
 - _set_core.sh_：用于方便调试设置系统生成 core dump 文件的脚本
 - _taosd-dump-cfg.gdb_：用于方便调试 taosd 的 gdb 执行脚本。

docs/zh/27-train-faq/03-docker.md

@@ -108,7 +108,7 @@ taos>
 也可以在宿主机使用 curl 通过 RESTful 端口访问 Docker 容器内的 TDengine server。
 
 ```
-curl -u root:taosdata -d 'show databases' 127.0.0.1:6041/rest/sql
+curl -L -u root:taosdata -d "show databases" 127.0.0.1:6041/rest/sql
 ```
 
 输出示例如下：
@@ -148,7 +148,7 @@ docker run -d --name tdengine-taosd -p 6030-6042:6030-6042 -p 6030-6042:6030-604
 使用 curl 命令验证 RESTful 接口可以正常工作：
 
 ```bash
-curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'show databases;' 127.0.0.1:6041/rest/sql
+curl -L -H "Authorization: Basic cm9vdDp0YW9zZGF0YQ==" -d "show databases;" 127.0.0.1:6041/rest/sql
 ```
 
 输出示例如下：

examples/c/stream_demo.c

@@ -25,19 +25,21 @@ int32_t init_env() {
     return -1;
   }
 
-  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
+  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1");
   if (taos_errno(pRes) != 0) {
     printf("error in create db, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
   taos_free_result(pRes);
 
+#if 0
   pRes = taos_query(pConn, "create database if not exists abc2 vgroups 20");
   if (taos_errno(pRes) != 0) {
     printf("error in create db, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
   taos_free_result(pRes);
+#endif
 
   pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -88,9 +90,9 @@ int32_t create_stream() {
   /*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/
   /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
-  pRes = taos_query(
+  pRes = taos_query(pConn,
-      pConn,
+                    "create stream stream1 trigger window_close into outstb as select _wstartts, sum(k) from st1 "
-      "create stream stream1 trigger at_once into abc1.outstb as select _wstartts, sum(k) from st1 interval(10m) ");
+                    "interval(10s) ");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;

include/common/systable.h

@@ -41,6 +41,7 @@ extern "C" {
 #define TSDB_INS_TABLE_VGROUPS                "vgroups"
 #define TSDB_INS_TABLE_VNODES                 "vnodes"
 #define TSDB_INS_TABLE_CONFIGS                "configs"
+#define TSDB_INS_TABLE_DNODE_VARIABLES        "dnode_variables"
 
 #define TSDB_PERFORMANCE_SCHEMA_DB     "performance_schema"
 #define TSDB_PERFS_TABLE_SMAS          "smas"

include/common/tmsg.h

@@ -1317,7 +1317,7 @@ int32_t tSerializeSSetStandbyReq(void* buf, int32_t bufLen, SSetStandbyReq* pReq
 int32_t tDeserializeSSetStandbyReq(void* buf, int32_t bufLen, SSetStandbyReq* pReq);
 
 typedef struct {
-  char    queryStrId[TSDB_QUERY_ID_LEN];
+  char queryStrId[TSDB_QUERY_ID_LEN];
 } SKillQueryReq;
 
 int32_t tSerializeSKillQueryReq(void* buf, int32_t bufLen, SKillQueryReq* pReq);
@@ -2282,6 +2282,9 @@ typedef struct {
   int8_t igNotExists;
 } SMDropStreamReq;
 
+int32_t tSerializeSMDropStreamReq(void* buf, int32_t bufLen, const SMDropStreamReq* pReq);
+int32_t tDeserializeSMDropStreamReq(void* buf, int32_t bufLen, SMDropStreamReq* pReq);
+
 typedef struct {
   int8_t reserved;
 } SMDropStreamRsp;

include/dnode/mnode/mnode.h

@@ -95,8 +95,8 @@ int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
  */
 int32_t mndProcessRpcMsg(SRpcMsg *pMsg);
 int32_t mndProcessSyncMsg(SRpcMsg *pMsg);
-int32_t mndPreProcessMsg(SRpcMsg *pMsg);
+int32_t mndPreProcessQueryMsg(SRpcMsg *pMsg);
-void mndAbortPreprocessMsg(SRpcMsg *pMsg);
+void    mndPostProcessQueryMsg(SRpcMsg *pMsg);
 
 /**
  * @brief Generate machine code

include/libs/catalog/catalog.h

@@ -69,6 +69,7 @@ typedef struct SCatalogReq {
   SArray* pUser;          // element is SUserAuthInfo
   SArray* pTableIndex;    // element is SNAME
   bool    qNodeRequired;  // valid qnode
+  bool    dNodeRequired;  // valid dnode
   bool    forceUpdate;
 } SCatalogReq;
 
@@ -88,6 +89,7 @@ typedef struct SMetaData {
   SArray* pIndex;       // pRes = SIndexInfo*
   SArray* pUser;        // pRes = bool*
   SArray* pQnodeList;   // pRes = SArray<SQueryNodeLoad>*
+  SArray* pDnodeList;   // pRes = SArray<SEpSet>*
 } SMetaData;
 
 typedef struct SCatalogCfg {
@@ -268,6 +270,8 @@ int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t
 
 int32_t catalogGetQnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray* pQnodeList);
 
+int32_t catalogGetDnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray** pDnodeList);
+
 int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableVersion **stables, uint32_t *num);
 
 int32_t catalogGetExpiredDBs(SCatalog* pCatalog, SDbVgVersion** dbs, uint32_t* num);

include/libs/function/function.h

@@ -67,7 +67,7 @@ typedef struct SResultRowEntryInfo {
   bool     initialized:1;     // output buffer has been initialized
   bool     complete:1;        // query has completed
   uint8_t  isNullRes:6;       // the result is null
-  uint8_t  numOfRes;          // num of output result in current buffer
+  uint16_t numOfRes;          // num of output result in current buffer
 } SResultRowEntryInfo;
 
 // determine the real data need to calculated the result

include/libs/nodes/cmdnodes.h

@@ -119,14 +119,14 @@ typedef struct SCreateTableStmt {
 } SCreateTableStmt;
 
 typedef struct SCreateSubTableClause {
-  ENodeType  type;
+  ENodeType      type;
-  char       dbName[TSDB_DB_NAME_LEN];
+  char           dbName[TSDB_DB_NAME_LEN];
-  char       tableName[TSDB_TABLE_NAME_LEN];
+  char           tableName[TSDB_TABLE_NAME_LEN];
-  char       useDbName[TSDB_DB_NAME_LEN];
+  char           useDbName[TSDB_DB_NAME_LEN];
-  char       useTableName[TSDB_TABLE_NAME_LEN];
+  char           useTableName[TSDB_TABLE_NAME_LEN];
-  bool       ignoreExists;
+  bool           ignoreExists;
-  SNodeList* pSpecificTags;
+  SNodeList*     pSpecificTags;
-  SNodeList* pValsOfTags;
+  SNodeList*     pValsOfTags;
   STableOptions* pOptions;
 } SCreateSubTableClause;
 
@@ -230,6 +230,11 @@ typedef struct SShowTableDistributedStmt {
   char      tableName[TSDB_TABLE_NAME_LEN];
 } SShowTableDistributedStmt;
 
+typedef struct SShowDnodeVariablesStmt {
+  ENodeType type;
+  SNode*    pDnodeId;
+} SShowDnodeVariablesStmt;
+
 typedef enum EIndexType { INDEX_TYPE_SMA = 1, INDEX_TYPE_FULLTEXT } EIndexType;
 
 typedef struct SIndexOptions {

include/libs/nodes/nodes.h

@@ -180,7 +180,9 @@ typedef enum ENodeType {
   QUERY_NODE_SHOW_VNODES_STMT,
   QUERY_NODE_SHOW_APPS_STMT,
   QUERY_NODE_SHOW_SCORES_STMT,
-  QUERY_NODE_SHOW_VARIABLE_STMT,
+  QUERY_NODE_SHOW_VARIABLES_STMT,
+  QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT,
+  QUERY_NODE_SHOW_DNODE_VARIABLES_STMT,
   QUERY_NODE_SHOW_CREATE_DATABASE_STMT,
   QUERY_NODE_SHOW_CREATE_TABLE_STMT,
   QUERY_NODE_SHOW_CREATE_STABLE_STMT,

include/util/taoserror.h

@@ -419,6 +419,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_SYN_NEW_CONFIG_ERROR          TAOS_DEF_ERROR_CODE(0, 0x090F)
 #define TSDB_CODE_SYN_RECONFIG_NOT_READY        TAOS_DEF_ERROR_CODE(0, 0x0910)
 #define TSDB_CODE_SYN_PROPOSE_NOT_READY         TAOS_DEF_ERROR_CODE(0, 0x0911)
+#define TSDB_CODE_SYN_STANDBY_NOT_READY         TAOS_DEF_ERROR_CODE(0, 0x0912)
 #define TSDB_CODE_SYN_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x09FF)
 
 // tq

packaging/tools/makepkg.sh

@@ -88,7 +88,6 @@ else
       ${build_dir}/bin/tarbitrator\
       ${script_dir}/remove.sh \
       ${script_dir}/set_core.sh \
-      ${script_dir}/run_taosd_and_taosadapter.sh \
       ${script_dir}/startPre.sh \
       ${script_dir}/taosd-dump-cfg.gdb"
 fi
@@ -152,7 +151,6 @@ if [ $adapterName != "taosadapter" ]; then
   sed -i "s/taosadapter/${adapterName}/g" ${install_dir}/cfg/$adapterName.service
 
   mv ${install_dir}/bin/taosadapter ${install_dir}/bin/${adapterName}
-  mv ${install_dir}/bin/run_taosd_and_taosadapter.sh ${install_dir}/bin/run_${serverName}_and_${adapterName}.sh
   mv ${install_dir}/bin/taosd-dump-cfg.gdb ${install_dir}/bin/${serverName}-dump-cfg.gdb
 fi
 

packaging/tools/run_taosd_and_taosadapter.sh

@@ -1,3 +0,0 @@
-#!/bin/bash
-[[ -x /usr/bin/taosadapter ]] && /usr/bin/taosadapter &
-taosd

source/client/src/clientMain.c

@@ -887,6 +887,7 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
 
   SRequestObj *pRequest = res;
   pRequest->body.fetchFp = fp;
+  pRequest->body.param = param;
 
   SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
   if (taos_num_fields(pRequest) == 0) {

source/client/test/clientTests.cpp

@@ -43,6 +43,26 @@ void showDB(TAOS* pConn) {
   }
 }
 
+void printResult(TAOS_RES* pRes) {
+  TAOS_ROW    pRow = NULL;
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  int32_t     numOfFields = taos_num_fields(pRes);
+
+  int32_t n = 0;
+  char str[512] = {0};
+  while ((pRow = taos_fetch_row(pRes)) != NULL) {
+    int32_t* length = taos_fetch_lengths(pRes);
+    for(int32_t i = 0; i < numOfFields; ++i) {
+      printf("(%d):%d " , i, length[i]);
+    }
+    printf("\n");
+
+    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+    printf("%s\n", str);
+    memset(str, 0, sizeof(str));
+  }
+}
+
 void fetchCallback(void* param, void* res, int32_t numOfRow) {
 #if 0
   printf("numOfRow = %d \n", numOfRow);
@@ -729,48 +749,31 @@ TEST(testCase, projection_query_tables) {
 //  taos_close(pConn);
 //}
 
-//TEST(testCase, agg_query_tables) {
-//  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-//  ASSERT_NE(pConn, nullptr);
-//
-//  TAOS_RES* pRes = taos_query(pConn, "use abc1");
-//  if (taos_errno(pRes) != 0) {
-//    printf("failed to use db, reason:%s\n", taos_errstr(pRes));
-//    taos_free_result(pRes);
-//    ASSERT_TRUE(false);
-//  }
-//  taos_free_result(pRes);
-//
-//  pRes = taos_query(pConn, "show stables");
-//  if (taos_errno(pRes) != 0) {
-//    printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
-//    taos_free_result(pRes);
-//    ASSERT_TRUE(false);
-//  }
-//
-//  TAOS_ROW    pRow = NULL;
-//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-//  int32_t     numOfFields = taos_num_fields(pRes);
-//
-//  int32_t n = 0;
-//  char str[512] = {0};
-//  while ((pRow = taos_fetch_row(pRes)) != NULL) {
-//    int32_t* length = taos_fetch_lengths(pRes);
-//    for(int32_t i = 0; i < numOfFields; ++i) {
-//      printf("(%d):%d " , i, length[i]);
-//    }
-//    printf("\n");
-//
-//    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
-//    printf("%s\n", str);
-//    memset(str, 0, sizeof(str));
-//  }
-//
-//  taos_free_result(pRes);
-//  taos_close(pConn);
-//}
-#endif
 
+TEST(testCase, agg_query_tables) {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  TAOS_RES* pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to use db, reason:%s\n", taos_errstr(pRes));
+    taos_free_result(pRes);
+    ASSERT_TRUE(false);
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "show table distributed st1");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
+    taos_free_result(pRes);
+    ASSERT_TRUE(false);
+  }
+
+  printResult(pRes);
+  taos_free_result(pRes);
+  taos_close(pConn);
+}
+#endif
 /*
 --- copy the following script in the shell to setup the environment ---
 
@@ -786,7 +789,7 @@ TEST(testCase, async_api_test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   ASSERT_NE(pConn, nullptr);
 
-  taos_query(pConn, "use table_alltype_hyperloglog");
+  taos_query(pConn, "use abc1");
 #if 0
   TAOS_RES* pRes = taos_query(pConn, "insert into tu(ts) values('2022-02-27 12:12:61')");
   if (taos_errno(pRes) != 0) {
@@ -812,7 +815,7 @@ TEST(testCase, async_api_test) {
   }
 #endif
 
-  taos_query_a(pConn, "select HYPERLOGLOG(q_ts) from stable_1_2 where ts between 1630000001000 and 1630100001000 interval(19d) Fill(NONE);", queryCallback, pConn);
+  taos_query_a(pConn, "select count(*) from tu", queryCallback, pConn);
   getchar();
   taos_close(pConn);
 }

source/common/src/tmsg.c

@@ -4267,6 +4267,35 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
   return 0;
 }
 
+int32_t tSerializeSMDropStreamReq(void *buf, int32_t bufLen, const SMDropStreamReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->igNotExists) < 0) return -1;
+
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSMDropStreamReq(void *buf, int32_t bufLen, SMDropStreamReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->igNotExists) < 0) return -1;
+
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
 void tFreeSCMCreateStreamReq(SCMCreateStreamReq *pReq) {
   taosMemoryFreeClear(pReq->sql);
   taosMemoryFreeClear(pReq->ast);

source/dnode/mgmt/mgmt_mnode/src/mmHandle.c

@@ -183,6 +183,7 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_SMA, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_SMA, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_STREAM, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_STREAM, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_GET_INDEX, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_GET_TABLE_INDEX, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

source/dnode/mgmt/mgmt_mnode/src/mmWorker.c

@@ -68,6 +68,10 @@ static void mmProcessRpcMsg(SQueueInfo *pInfo, SRpcMsg *pMsg) {
     mmSendRsp(pMsg, code);
   }
 
+  if (code == TSDB_CODE_RPC_REDIRECT) {
+    mndPostProcessQueryMsg(pMsg);
+  }
+
   dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
   rpcFreeCont(pMsg->pCont);
   taosFreeQitem(pMsg);
@@ -116,7 +120,7 @@ int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
 int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   pMsg->info.node = pMgmt->pMnode;
-  if (mndPreProcessMsg(pMsg) != 0) {
+  if (mndPreProcessQueryMsg(pMsg) != 0) {
     dError("msg:%p, failed to pre-process in mnode since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pMsg->msgType));
     return -1;
   }

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

@@ -550,12 +550,10 @@ static int32_t mndCheckMnodeState(SRpcMsg *pMsg) {
   if (!IsReq(pMsg)) return 0;
   if (mndAcquireRpcRef(pMsg->info.node) == 0) return 0;
 
-  if (IsReq(pMsg) && pMsg->msgType != TDMT_MND_MQ_TIMER && pMsg->msgType != TDMT_MND_TELEM_TIMER &&
+  if (pMsg->msgType != TDMT_MND_MQ_TIMER && pMsg->msgType != TDMT_MND_TELEM_TIMER &&
       pMsg->msgType != TDMT_MND_TRANS_TIMER) {
     mError("msg:%p, failed to check mnode state since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pMsg->msgType));
 
-    mndAbortPreprocessMsg(pMsg);
-
     SEpSet epSet = {0};
     mndGetMnodeEpSet(pMsg->info.node, &epSet);
 

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

@@ -18,16 +18,14 @@
 #include "mndMnode.h"
 #include "qworker.h"
 
-int32_t mndPreProcessMsg(SRpcMsg *pMsg) {
+int32_t mndPreProcessQueryMsg(SRpcMsg *pMsg) {
   if (TDMT_VND_QUERY != pMsg->msgType) return 0;
-
   SMnode *pMnode = pMsg->info.node;
   return qWorkerPreprocessQueryMsg(pMnode->pQuery, pMsg);
 }
 
-void mndAbortPreprocessMsg(SRpcMsg *pMsg) {
+void mndPostProcessQueryMsg(SRpcMsg *pMsg) {
   if (TDMT_VND_QUERY != pMsg->msgType) return;
-
   SMnode *pMnode = pMsg->info.node;
   qWorkerAbortPreprocessQueryMsg(pMnode->pQuery, pMsg);
 }

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

@@ -496,6 +496,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       // input
       pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK;
 
+      // trigger
+      pTask->triggerParam = pStream->triggerParam;
+
       // sink or dispatch
       if (hasExtraSink) {
         mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pTask);

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

@@ -408,6 +408,7 @@ static int32_t mndProcessDropSnodeReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
+  // check deletable
   code = mndDropSnode(pMnode, pReq, pObj);
   if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
 

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

@@ -518,7 +518,6 @@ static int32_t mndCreateStream(SMnode *pMnode, SRpcMsg *pReq, SCMCreateStreamReq
   // TODO
   streamObj.fixedSinkVgId = 0;
   streamObj.smaId = 0;
-  /*streamObj.physicalPlan = "";*/
   streamObj.trigger = pCreate->triggerType;
   streamObj.watermark = pCreate->watermark;
   streamObj.triggerParam = pCreate->maxDelay;
@@ -607,17 +606,6 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
   }
 #endif
 
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq);
-  if (pTrans == NULL) {
-    mError("stream:%s, failed to create since %s", createStreamReq.name, terrstr());
-    goto _OVER;
-  }
-
-  mndTransSetDbName(pTrans, createStreamReq.sourceDB, NULL);
-  // TODO
-  /*mndTransSetDbName(pTrans, streamObj.targetDb, NULL);*/
-  mDebug("trans:%d, used to create stream:%s", pTrans->id, createStreamReq.name);
-
   // build stream obj from request
   SStreamObj streamObj = {0};
   if (mndBuildStreamObjFromCreateReq(pMnode, &streamObj, &createStreamReq) < 0) {
@@ -626,6 +614,14 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq);
+  if (pTrans == NULL) {
+    mError("stream:%s, failed to create since %s", createStreamReq.name, terrstr());
+    goto _OVER;
+  }
+  mndTransSetDbName(pTrans, createStreamReq.sourceDB, streamObj.targetDb);
+  mDebug("trans:%d, used to create stream:%s", pTrans->id, createStreamReq.name);
+
   // create stb for stream
   if (mndCreateStbForStream(pMnode, pTrans, &streamObj, pReq->info.conn.user) < 0) {
     mError("trans:%d, failed to create stb for stream %s since %s", pTrans->id, createStreamReq.name, terrstr());

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

@@ -75,12 +75,10 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM
 }
 
 int32_t mndSyncGetSnapshot(struct SSyncFSM *pFsm, SSnapshot *pSnapshot, void *pReaderParam, void **ppReader) {
-  // TODO:
+  mInfo("start to read snapshot from sdb in atomic way");
-
+  SMnode *pMnode = pFsm->data;
-  // atomic operation
+  return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader, &pSnapshot->lastApplyIndex, &pSnapshot->lastApplyTerm,
-  // step1. sdbGetCommitInfo
+                      &pSnapshot->lastConfigIndex);
-  // step2. create ppReader with pReaderParam
-
   return 0;
 }
 
@@ -106,14 +104,6 @@ void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta cbM
   SMnode    *pMnode = pFsm->data;
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
 
-#if 0
-// send response
-  SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen, .conn.applyIndex = cbMeta.index};
-  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
-  memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
-  syncGetAndDelRespRpc(pMnode->syncMgmt.sync, cbMeta.seqNum, &rpcMsg.info);
-#endif
-
   pMgmt->errCode = cbMeta.code;
   mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64, pMgmt->transId,
         cbMeta.code, cbMeta.index, cbMeta.term);
@@ -130,7 +120,7 @@ void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta cbM
 int32_t mndSnapshotStartRead(struct SSyncFSM *pFsm, void **ppReader) {
   mInfo("start to read snapshot from sdb");
   SMnode *pMnode = pFsm->data;
-  return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader);
+  return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader, NULL, NULL, NULL);
 }
 
 int32_t mndSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) {

source/dnode/mnode/impl/test/sdb/sdbTest.cpp

@@ -910,7 +910,7 @@ TEST_F(MndTestSdb, 01_Read_Str) {
     int32_t   len = 0;
     int32_t   code = 0;
 
-    code = sdbStartRead(pSdb, &pReader);
+    code = sdbStartRead(pSdb, &pReader, NULL, NULL, NULL);
     ASSERT_EQ(code, 0);
     code = sdbStartWrite(pSdb, &pWritter);
     ASSERT_EQ(code, 0);

source/dnode/mnode/sdb/inc/sdb.h

@@ -388,7 +388,7 @@ SSdbRow *sdbAllocRow(int32_t objSize);
 void    *sdbGetRowObj(SSdbRow *pRow);
 void     sdbFreeRow(SSdb *pSdb, SSdbRow *pRow, bool callFunc);
 
-int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter);
+int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter, int64_t *index, int64_t *term, int64_t *config);
 int32_t sdbStopRead(SSdb *pSdb, SSdbIter *pIter);
 int32_t sdbDoRead(SSdb *pSdb, SSdbIter *pIter, void **ppBuf, int32_t *len);
 

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

@@ -523,7 +523,7 @@ static void sdbCloseIter(SSdbIter *pIter) {
   taosMemoryFree(pIter);
 }
 
-int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter) {
+int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter, int64_t *index, int64_t *term, int64_t *config) {
   SSdbIter *pIter = sdbCreateIter(pSdb);
   if (pIter == NULL) return -1;
 
@@ -552,6 +552,10 @@ int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter) {
   }
 
   *ppIter = pIter;
+  if (index != NULL) *index = commitIndex;
+  if (term != NULL) *term = commitTerm;
+  if (config != NULL) *config = commitConfig;
+
   mInfo("sdbiter:%p, is created to read snapshot, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " file:%s",
         pIter, commitIndex, commitTerm, commitConfig, pIter->name);
   return 0;

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

@@ -336,7 +336,7 @@ static void vnodeSyncReconfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReCon
   SVnode *pVnode = pFsm->data;
 
   SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
-  syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info);
+  syncGetAndDelRespRpc(pVnode->sync, cbMeta.newCfgSeqNum, &rpcMsg.info);
   rpcMsg.info.conn.applyIndex = cbMeta.index;
 
   STraceId *trace = (STraceId *)&pMsg->info.traceId;

source/libs/catalog/src/catalog.c

@@ -1121,6 +1121,10 @@ _return:
   CTG_API_LEAVE(TSDB_CODE_SUCCESS);
 }
 
+int32_t catalogGetDnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray** pDnodeList) {
+  return TSDB_CODE_CTG_INVALID_INPUT;
+}
+
 int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableVersion **stables, uint32_t *num) {
   CTG_API_ENTER();
 

source/libs/command/src/command.c

@@ -131,6 +131,8 @@ static int32_t execShowCreateSTable(SShowCreateTableStmt* pStmt) { return TSDB_C
 
 static int32_t execAlterLocal(SAlterLocalStmt* pStmt) { return TSDB_CODE_FAILED; }
 
+static int32_t execShowLocalVariables() { return TSDB_CODE_FAILED; }
+
 int32_t qExecCommand(SNode* pStmt, SRetrieveTableRsp** pRsp) {
   switch (nodeType(pStmt)) {
     case QUERY_NODE_DESCRIBE_STMT:
@@ -145,6 +147,8 @@ int32_t qExecCommand(SNode* pStmt, SRetrieveTableRsp** pRsp) {
       return execShowCreateSTable((SShowCreateTableStmt*)pStmt);
     case QUERY_NODE_ALTER_LOCAL_STMT:
       return execAlterLocal((SAlterLocalStmt*)pStmt);
+    case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
+      return execShowLocalVariables();
     default:
       break;
   }

source/libs/executor/inc/executil.h

@@ -123,4 +123,6 @@ SColumn   extractColumnFromColumnNode(SColumnNode* pColNode);
 int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode);
 void    cleanupQueryTableDataCond(SQueryTableDataCond* pCond);
 
+int32_t convertFillType(int32_t mode);
+
 #endif  // TDENGINE_QUERYUTIL_H

source/libs/executor/inc/executorimpl.h

@@ -566,13 +566,14 @@ typedef struct SStreamSessionAggOperatorInfo {
 } SStreamSessionAggOperatorInfo;
 
 typedef struct STimeSliceOperatorInfo {
-  SOptrBasicInfo binfo;
+  SSDataBlock*   pRes;
   STimeWindow    win;
   SInterval      interval;
   int64_t        current;
   SArray*        pPrevRow;      // SArray<SGroupValue>
-  SArray*        pCols;         // SArray<SColumn>
   int32_t        fillType;      // fill type
+  SColumn        tsCol;         // primary timestamp column
+  SExprSupp      scalarSup;     // scalar calculation
   struct SFillColInfo*  pFillColInfo;  // fill column info
 } STimeSliceOperatorInfo;
 
@@ -670,7 +671,7 @@ int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t
 
 void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
 void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
-void    initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
+int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
 void    cleanupExprSup(SExprSupp* pSup);
 int32_t initAggInfo(SExprSupp *pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                     const char* pkey);
@@ -757,8 +758,8 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf
 
 SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
+SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode, /*SExprInfo* pExprInfo, int32_t numOfCols,
-                                           SSDataBlock* pResultBlock, const SNodeListNode* pValNode, SExecTaskInfo* pTaskInfo);
+                                           SSDataBlock* pResultBlock, const SNodeListNode* pValNode, */SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SJoinPhysiNode* pJoinNode,
                                            SExecTaskInfo* pTaskInfo);

source/libs/executor/src/executil.c

@@ -694,4 +694,32 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
 void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {
   taosMemoryFree(pCond->twindows);
   taosMemoryFree(pCond->colList);
-}
+}
+
+int32_t convertFillType(int32_t mode) {
+  int32_t type = TSDB_FILL_NONE;
+  switch (mode) {
+    case FILL_MODE_PREV:
+      type = TSDB_FILL_PREV;
+      break;
+    case FILL_MODE_NONE:
+      type = TSDB_FILL_NONE;
+      break;
+    case FILL_MODE_NULL:
+      type = TSDB_FILL_NULL;
+      break;
+    case FILL_MODE_NEXT:
+      type = TSDB_FILL_NEXT;
+      break;
+    case FILL_MODE_VALUE:
+      type = TSDB_FILL_SET_VALUE;
+      break;
+    case FILL_MODE_LINEAR:
+      type = TSDB_FILL_LINEAR;
+      break;
+    default:
+      type = TSDB_FILL_NONE;
+  }
+
+  return type;
+}

source/libs/executor/src/executorimpl.c

@@ -2754,7 +2754,11 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t
     goto _error;
   }
 
-  pOperator->exprSupp.pCtx = createSqlFunctionCtx(pExprInfo, num, &pOperator->exprSupp.rowEntryInfoOffset);
+  int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, num);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _error;
+  }
+
   initResultRowInfo(&pInfo->binfo.resultRowInfo);
 
   if (pOperator->exprSupp.pCtx == NULL || pInfo->binfo.pRes == NULL) {
@@ -2762,7 +2766,7 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t
   }
 
   size_t  keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
-  int32_t code = doInitAggInfoSup(&pInfo->aggSup, pOperator->exprSupp.pCtx, num, keyBufSize, pTaskInfo->id.str);
+  code = doInitAggInfoSup(&pInfo->aggSup, pOperator->exprSupp.pCtx, num, keyBufSize, pTaskInfo->id.str);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -2783,12 +2787,9 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t
 
   pOperator->name = "SortedMerge";
   // pOperator->operatorType = OP_SortedMerge;
-  pOperator->blocking = true;
+  pOperator->blocking  = true;
-  pOperator->status = OP_NOT_OPENED;
+  pOperator->status    = OP_NOT_OPENED;
-  pOperator->info = pInfo;
+  pOperator->info      = pInfo;
-  pOperator->exprSupp.numOfExprs = num;
-  pOperator->exprSupp.pExprInfo = pExprInfo;
-
   pOperator->pTaskInfo = pTaskInfo;
 
   pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSortedMerge, NULL, NULL, destroySortedMergeOperatorInfo,
@@ -3408,7 +3409,11 @@ void cleanupAggSup(SAggSupporter* pAggSup) {
 
 int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                     const char* pkey) {
-  initExprSupp(pSup, pExprInfo, numOfCols);
+  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
+
   doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey);
   for (int32_t i = 0; i < numOfCols; ++i) {
     pSup->pCtx[i].pBuf = pAggSup->pResultBuf;
@@ -3431,12 +3436,17 @@ void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock) {
   initResultRowInfo(&pInfo->resultRowInfo);
 }
 
-void initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr) {
+int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr) {
   pSup->pExprInfo = pExprInfo;
   pSup->numOfExprs = numOfExpr;
   if (pSup->pExprInfo != NULL) {
     pSup->pCtx = createSqlFunctionCtx(pExprInfo, numOfExpr, &pSup->rowEntryInfoOffset);
+    if (pSup->pCtx == NULL) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
@@ -3458,7 +3468,10 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo*
   }
 
   initBasicInfo(&pInfo->binfo, pResultBlock);
-  initExprSupp(&pInfo->scalarExprSup, pScalarExprInfo, numOfScalarExpr);
+  code = initExprSupp(&pInfo->scalarExprSup, pScalarExprInfo, numOfScalarExpr);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _error;
+  }
 
   pInfo->groupId = INT32_MIN;
   pOperator->name = "TableAggregate";
@@ -3723,13 +3736,15 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy
   SExprInfo* pExprInfo = createExprInfo(pPhyNode->pFuncs, NULL, &numOfExpr);
 
   if (pPhyNode->pExprs != NULL) {
-    SExprSupp* pSup1 = &pInfo->scalarSup;
+    int32_t num = 0;
-    pSup1->pExprInfo = createExprInfo(pPhyNode->pExprs, NULL, &pSup1->numOfExprs);
+    SExprInfo* pSExpr = createExprInfo(pPhyNode->pExprs, NULL, &num);
-    pSup1->pCtx = createSqlFunctionCtx(pSup1->pExprInfo, pSup1->numOfExprs, &pSup1->rowEntryInfoOffset);
+    int32_t code = initExprSupp(&pInfo->scalarSup, pSExpr, num);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _error;
+    }
   }
 
   SSDataBlock* pResBlock = createResDataBlock(pPhyNode->node.pOutputDataBlockDesc);
-  ;
 
   int32_t numOfRows = 4096;
   size_t  keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
@@ -3746,15 +3761,14 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy
 
   setFunctionResultOutput(pOperator, &pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfExpr);
 
-  pInfo->binfo.pRes = pResBlock;
+  pInfo->binfo.pRes       = pResBlock;
-  pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pSup->pCtx, numOfExpr);
+  pInfo->pPseudoColInfo   = setRowTsColumnOutputInfo(pSup->pCtx, numOfExpr);
 
   pOperator->name = "IndefinitOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT;
-  pOperator->blocking = false;
+  pOperator->blocking     = false;
-  pOperator->status = OP_NOT_OPENED;
+  pOperator->status       = OP_NOT_OPENED;
-  pOperator->info = pInfo;
+  pOperator->info         = pInfo;
-  pOperator->exprSupp.pExprInfo = pExprInfo;
   pOperator->exprSupp.numOfExprs = numOfExpr;
   pOperator->pTaskInfo = pTaskInfo;
 
@@ -3795,34 +3809,6 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t
   }
 }
 
-static int32_t convertFillType(int32_t mode) {
-  int32_t type = TSDB_FILL_NONE;
-  switch (mode) {
-    case FILL_MODE_PREV:
-      type = TSDB_FILL_PREV;
-      break;
-    case FILL_MODE_NONE:
-      type = TSDB_FILL_NONE;
-      break;
-    case FILL_MODE_NULL:
-      type = TSDB_FILL_NULL;
-      break;
-    case FILL_MODE_NEXT:
-      type = TSDB_FILL_NEXT;
-      break;
-    case FILL_MODE_VALUE:
-      type = TSDB_FILL_SET_VALUE;
-      break;
-    case FILL_MODE_LINEAR:
-      type = TSDB_FILL_LINEAR;
-      break;
-    default:
-      type = TSDB_FILL_NONE;
-  }
-
-  return type;
-}
-
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, bool multigroupResult,
                                       SExecTaskInfo* pTaskInfo) {
   SFillOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SFillOperatorInfo));
@@ -3852,10 +3838,10 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
   pOperator->blocking = false;
   pOperator->status = OP_NOT_OPENED;
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL;
-  pOperator->exprSupp.pExprInfo = pExprInfo;
+  pOperator->exprSupp.pExprInfo   = pExprInfo;
-  pOperator->exprSupp.numOfExprs = num;
+  pOperator->exprSupp.numOfExprs  = num;
-  pOperator->info = pInfo;
+  pOperator->info         = pInfo;
-  pOperator->pTaskInfo = pTaskInfo;
+  pOperator->pTaskInfo    = pTaskInfo;
 
   pOperator->fpSet =
       createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, NULL, destroySFillOperatorInfo, NULL, NULL, NULL);
@@ -4282,6 +4268,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     pOptr = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, false, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) {
     pOptr = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo);
+  } else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) {
+    pOptr = createTimeSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo);
   } else {
     ASSERT(0);
   }

source/libs/executor/src/groupoperator.c

@@ -387,11 +387,12 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
   pInfo->pGroupCols      = pGroupColList;
   pInfo->pCondition      = pCondition;
 
-  pInfo->scalarSup.pExprInfo = pScalarExprInfo;
+  int32_t code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr);
-  pInfo->scalarSup.numOfExprs = numOfScalarExpr;
+  if (code != TSDB_CODE_SUCCESS) {
-  pInfo->scalarSup.pCtx  = createSqlFunctionCtx(pScalarExprInfo, numOfScalarExpr, &pInfo->scalarSup.rowEntryInfoOffset);
+    goto _error;
+  }
 
-  int32_t code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pGroupColList);
+  code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pGroupColList);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -718,10 +719,12 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
   pInfo->pGroupCols = extractPartitionColInfo(pPartNode->pPartitionKeys);
 
   if (pPartNode->pExprs != NULL) {
-    pInfo->scalarSup.numOfExprs = 0;
+    int32_t num = 0;
-    pInfo->scalarSup.pExprInfo = createExprInfo(pPartNode->pExprs, NULL, &pInfo->scalarSup.numOfExprs);
+    SExprInfo* pExprInfo1 = createExprInfo(pPartNode->pExprs, NULL, &num);
-    pInfo->scalarSup.pCtx = createSqlFunctionCtx(
+    int32_t code = initExprSupp(&pInfo->scalarSup, pExprInfo1, num);
-        pInfo->scalarSup.pExprInfo, pInfo->scalarSup.numOfExprs, &pInfo->scalarSup.rowEntryInfoOffset);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _error;
+    }
   }
 
   _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

source/libs/executor/src/scanoperator.c

@@ -686,7 +686,10 @@ SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SReadHandle* re
 
   int32_t numOfCols = 0;
   SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols);
-  initExprSupp(&pOperator->exprSupp, pExprInfo, numOfCols);
+  int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfCols);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _error;
+  }
 
   pOperator->name      = "DataBlockDistScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN;
@@ -1872,7 +1875,10 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
   SArray*    colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID);
 
 
-  initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs);
+  int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _error;
+  }
 
   pInfo->pTableList       = pTableListInfo;
   pInfo->pColMatchInfo    = colList;

source/libs/executor/src/timewindowoperator.c

@@ -1705,7 +1705,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
   return (rows == 0) ? NULL : pBInfo->pRes;
 }
 
-static void doKeepPrevRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock* pBlock) {
+static void doKeepPrevRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock* pBlock, int32_t rowIndex) {
   int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
   for(int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData*  pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
@@ -1715,108 +1715,53 @@ static void doKeepPrevRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock
       SGroupKeys* pkey = taosArrayGet(pSliceInfo->pPrevRow, i);
 
       pkey->isNull = false;
-      char* val = colDataGetData(pColInfoData, i);
+      char* val = colDataGetData(pColInfoData, rowIndex);
       memcpy(pkey->pData, val, pkey->bytes);
     }
   }
 }
 
-static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
+static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pBlock,
-  if (pOperator->status == OP_EXEC_DONE) {
+                                   int32_t rowIndex, SSDataBlock* pResBlock) {
-    return NULL;
+  int32_t rows = pResBlock->info.rows;
-  }
 
-  STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
+  // todo set the correct primary timestamp column
-  SSDataBlock* pResBlock = pSliceInfo->binfo.pRes;
-  SExprSupp* pSup = &pOperator->exprSupp;
 
-//  if (pOperator->status == OP_RES_TO_RETURN) {
+  // output the result
-//    //    doBuildResultDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pIntervalInfo->pRes);
+  for (int32_t j = 0; j < pExprSup->numOfExprs; ++j) {
-//    if (pResBlock->info.rows == 0 || !hasDataInGroupInfo(&pSliceInfo->groupResInfo)) {
+    SExprInfo* pExprInfo = &pExprSup->pExprInfo[j];
-//      doSetOperatorCompleted(pOperator);
+    int32_t    dstSlot = pExprInfo->base.resSchema.slotId;
-//    }
+    int32_t    srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
-//
-//    return pResBlock;
-//  }
 
-  int32_t order = TSDB_ORDER_ASC;
+    SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
-  SInterval* pInterval = &pSliceInfo->interval;
+    SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
-  SOperatorInfo* downstream = pOperator->pDownstream[0];
 
-  int32_t numOfRows = 0;
+    switch (pSliceInfo->fillType) {
-  while (1) {
+      case TSDB_FILL_NULL:
-    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
+        colDataAppendNULL(pDst, rows);
-    if (pBlock == NULL) {
+        break;
-      break;
-    }
 
-    // the pDataBlock are always the same one, no need to call this again
+      case TSDB_FILL_SET_VALUE: {
-    setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
+        SVariant* pVar = &pSliceInfo->pFillColInfo[j].fillVal;
 
-    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, 0);
+        if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
-    for(int32_t i = 0; i < pBlock->info.rows; ++i) {
+          float v = 0;
-      int64_t ts = *(int64_t*) colDataGetData(pTsCol, i);
+          GET_TYPED_DATA(v, float, pVar->nType, &pVar->i);
-
+          colDataAppend(pDst, rows, (char*)&v, false);
-      if (ts == pSliceInfo->current) {
+        } else if (pDst->info.type == TSDB_DATA_TYPE_DOUBLE) {
-        for(int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
+          double v = 0;
-          SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j];
+          GET_TYPED_DATA(v, double, pVar->nType, &pVar->i);
-          int32_t dstSlot = pExprInfo->base.resSchema.slotId;
+          colDataAppend(pDst, rows, (char*)&v, false);
-          int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
+        } else if (IS_SIGNED_NUMERIC_TYPE(pDst->info.type)) {
-
+          int64_t v = 0;
-          SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
+          GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);
-          SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, dstSlot);
+          colDataAppend(pDst, rows, (char*)&v, false);
-
-          char* v = colDataGetData(pSrc, i);
-          colDataAppend(pDst, numOfRows, v, false);
         }
+      } break;
 
-        numOfRows += 1;
+      case TSDB_FILL_LINEAR:
-
-        pSliceInfo->current += taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
-        if (pSliceInfo->current > pSliceInfo->win.ekey) {
-          doSetOperatorCompleted(pOperator);
-          break;
-        }
-      } else if (ts < pSliceInfo->current) {
-        if (i != pBlock->info.window.ekey) {
-          int64_t nextTs = *(int64_t*) colDataGetData(pTsCol, i + 1);
-          if (nextTs > pSliceInfo->current) {
-            // output the result
-            for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
-              SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j];
-              int32_t    dstSlot = pExprInfo->base.resSchema.slotId;
-              int32_t    srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
-
-              SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
-              SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, dstSlot);
-
-              switch (pSliceInfo->fillType) {
-                case TSDB_FILL_NULL:
-                  colDataAppendNULL(pDst, numOfRows);
-                  break;
-
-                case TSDB_FILL_SET_VALUE: {
-                  SVariant* pVar = &pSliceInfo->pFillColInfo[i].fillVal;
-
-                  if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
-                    float v = 0;
-                    GET_TYPED_DATA(v, float, pVar->nType, &pVar->i);
-                    colDataAppend(pDst, numOfRows, (char*)&v, false);
-                  } else if (pDst->info.type == TSDB_DATA_TYPE_DOUBLE) {
-                    double v = 0;
-                    GET_TYPED_DATA(v, double, pVar->nType, &pVar->i);
-                    colDataAppend(pDst, numOfRows, (char*)&v, false);
-                  } else if (IS_SIGNED_NUMERIC_TYPE(pDst->info.type)) {
-                    int64_t v = 0;
-                    GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);
-                    colDataAppend(pDst, numOfRows, (char*)&v, false);
-                  }
-                }
-                break;
-
-                case TSDB_FILL_LINEAR:
 #if 0
-                if (pCtx->start.key == INT64_MIN || pCtx->start.key > pCtx->startTs
+        if (pCtx->start.key == INT64_MIN || pCtx->start.key > pCtx->startTs
                     || pCtx->end.key == INT64_MIN || pCtx->end.key < pCtx->startTs) {
 //                  goto interp_exit;
                 }
@@ -1845,33 +1790,158 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
                 }
               }
 #endif
-                  break;
+        break;
 
-                case TSDB_FILL_PREV: {
+      case TSDB_FILL_PREV: {
-                  SGroupKeys* pkey = taosArrayGet(pSliceInfo->pPrevRow, srcSlot);
+        SGroupKeys* pkey = taosArrayGet(pSliceInfo->pPrevRow, srcSlot);
-                  colDataAppend(pDst, numOfRows, pkey->pData, false);
+        colDataAppend(pDst, rows, pkey->pData, false);
-                } break;
+      } break;
 
-                case TSDB_FILL_NEXT: {
+      case TSDB_FILL_NEXT: {
-                } break;
+        char* p = colDataGetData(pSrc, rowIndex);
+        colDataAppend(pDst, rows, p, colDataIsNull_s(pSrc, rowIndex));
+      } break;
 
-                case TSDB_FILL_NONE:
+      case TSDB_FILL_NONE:
-                default:
+      default:
-                  break;
+        break;
-              }
+    }
+  }
 
-              pSliceInfo->current +=
+  pResBlock->info.rows += 1;
+}
+
+static int32_t initPrevRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pBlock) {
+  if (pInfo->pPrevRow != NULL) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  pInfo->pPrevRow = taosArrayInit(4, sizeof(SGroupKeys));
+  if (pInfo->pPrevRow == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  int32_t numOfCols = pBlock->info.numOfCols;
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
+
+    SGroupKeys key = {0};
+    key.bytes = pColInfo->info.bytes;
+    key.type = pColInfo->info.type;
+    key.isNull = false;
+    key.pData = taosMemoryCalloc(1, pColInfo->info.bytes);
+    taosArrayPush(pInfo->pPrevRow, &key);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
+  if (pOperator->status == OP_EXEC_DONE) {
+    return NULL;
+  }
+
+  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
+
+  STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
+  SSDataBlock* pResBlock = pSliceInfo->pRes;
+  SExprSupp* pSup = &pOperator->exprSupp;
+
+  blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);
+
+//  if (pOperator->status == OP_RES_TO_RETURN) {
+//    //    doBuildResultDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pIntervalInfo->pRes);
+//    if (pResBlock->info.rows == 0 || !hasDataInGroupInfo(&pSliceInfo->groupResInfo)) {
+//      doSetOperatorCompleted(pOperator);
+//    }
+//
+//    return pResBlock;
+//  }
+
+  int32_t order = TSDB_ORDER_ASC;
+  SInterval* pInterval = &pSliceInfo->interval;
+  SOperatorInfo* downstream = pOperator->pDownstream[0];
+
+  int32_t numOfRows = 0;
+  while (1) {
+    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
+    if (pBlock == NULL) {
+      break;
+    }
+
+    int32_t code = initPrevRowsKeeper(pSliceInfo, pBlock);
+    if (code != TSDB_CODE_SUCCESS) {
+      longjmp(pTaskInfo->env, code);
+    }
+
+    // the pDataBlock are always the same one, no need to call this again
+    setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
+
+    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
+    for(int32_t i = 0; i < pBlock->info.rows; ++i) {
+      int64_t ts = *(int64_t*) colDataGetData(pTsCol, i);
+
+      if (ts == pSliceInfo->current) {
+        for(int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
+          SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j];
+          int32_t dstSlot = pExprInfo->base.resSchema.slotId;
+          int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
+
+          SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
+          SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
+
+          char* v = colDataGetData(pSrc, i);
+          colDataAppend(pDst, numOfRows, v, false);
+        }
+
+        pResBlock->info.rows += 1;
+        doKeepPrevRows(pSliceInfo, pBlock, i);
+
+        pSliceInfo->current = taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+        if (pSliceInfo->current > pSliceInfo->win.ekey) {
+          doSetOperatorCompleted(pOperator);
+          break;
+        }
+
+        if (pResBlock->info.rows >= pResBlock->info.capacity) {
+          break;
+        }
+      } else if (ts < pSliceInfo->current) {
+        if (i < pBlock->info.rows - 1) {
+          int64_t nextTs = *(int64_t*) colDataGetData(pTsCol, i + 1);
+          if (nextTs > pSliceInfo->current) {
+            while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
+              genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
+              pSliceInfo->current =
                   taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
-              if (pSliceInfo->current > pSliceInfo->win.ekey) {
+              if (pResBlock->info.rows >= pResBlock->info.capacity) {
-                doSetOperatorCompleted(pOperator);
                 break;
               }
             }
+
+            if (pSliceInfo->current > pSliceInfo->win.ekey) {
+              doSetOperatorCompleted(pOperator);
+              break;
+            }
           } else {
             // ignore current row, and do nothing
           }
         } else {  // it is the last row of current block
-          doKeepPrevRows(pSliceInfo, pBlock);
+          doKeepPrevRows(pSliceInfo, pBlock, i);
+        }
+      } else {  // ts > pSliceInfo->current
+        while (pSliceInfo->current < ts && pSliceInfo->current <= pSliceInfo->win.ekey) {
+          genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
+          pSliceInfo->current =
+              taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
+          if (pResBlock->info.rows >= pResBlock->info.capacity) {
+            break;
+          }
+        }
+
+        if (pSliceInfo->current > pSliceInfo->win.ekey) {
+          doSetOperatorCompleted(pOperator);
+          break;
         }
       }
     }
@@ -1886,59 +1956,46 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
   return pResBlock->info.rows == 0 ? NULL : pResBlock;
 }
 
-static int32_t initTimesliceInfo(STimeSliceOperatorInfo* pInfo, SqlFunctionCtx* pCtx, int32_t numOfCols) {
+SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode *pPhyNode, SExecTaskInfo* pTaskInfo) {
-  pInfo->pPrevRow = taosArrayInit(4, sizeof(SGroupKeys));
-  pInfo->pCols = taosArrayInit(4, sizeof(SColumn));
-
-  if (pInfo->pPrevRow == NULL || pInfo->pCols == NULL) {
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-
-  for (int32_t i = 0; i < numOfCols; ++i) {
-    SExprInfo* pExpr = pCtx[i].pExpr;
-
-    SFunctParam* pParam = &pExpr->base.pParam[0];
-
-    SColumn c = *pParam->pCol;
-    taosArrayPush(pInfo->pCols, &c);
-
-    SGroupKeys key = {0};
-    key.bytes = c.bytes;
-    key.type = c.type;
-    key.isNull = false;
-    key.pData = taosMemoryCalloc(1, c.bytes);
-    taosArrayPush(pInfo->pPrevRow, &key);
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-
-SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
-                                           SSDataBlock* pResultBlock, const SNodeListNode* pValNode, SExecTaskInfo* pTaskInfo) {
   STimeSliceOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STimeSliceOperatorInfo));
   SOperatorInfo*          pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pOperator == NULL || pInfo == NULL) {
     goto _error;
   }
 
+  SInterpFuncPhysiNode* pInterpPhyNode = (SInterpFuncPhysiNode*)pPhyNode;
   SExprSupp* pSup = &pOperator->exprSupp;
 
-  int32_t code = initTimesliceInfo(pInfo, pSup->pCtx, numOfCols);
+  int32_t numOfExprs = 0;
+  SExprInfo* pExprInfo = createExprInfo(pInterpPhyNode->pFuncs, NULL, &numOfExprs);
+  int32_t code = initExprSupp(pSup, pExprInfo, numOfExprs);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
 
-  initResultRowInfo(&pInfo->binfo.resultRowInfo);
+  if (pInterpPhyNode->pExprs != NULL) {
-  pInfo->pFillColInfo = createFillColInfo(pExprInfo, numOfCols, pValNode);
+    int32_t num = 0;
+    SExprInfo* pScalarExprInfo = createExprInfo(pInterpPhyNode->pExprs, NULL, &num);
+    code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, num);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _error;
+    }
+  }
 
-  pInfo->binfo.pRes     = pResultBlock;
+  pInfo->tsCol = extractColumnFromColumnNode((SColumnNode*)pInterpPhyNode->pTimeSeries);
+  pInfo->fillType = convertFillType(pInterpPhyNode->fillMode);
+  initResultSizeInfo(pOperator, 4096);
+
+  pInfo->pFillColInfo   = createFillColInfo(pExprInfo, numOfExprs, (SNodeListNode*)pInterpPhyNode->pFillValues);
+  pInfo->pRes           = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
+  pInfo->win            = pInterpPhyNode->timeRange;
+  pInfo->interval.interval = pInterpPhyNode->interval;
+  pInfo->current        = pInfo->win.skey;
 
   pOperator->name       = "TimeSliceOperator";
-  //  pOperator->operatorType = OP_AllTimeWindow;
+  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC;
-  pOperator->blocking   = true;
+  pOperator->blocking   = false;
   pOperator->status     = OP_NOT_OPENED;
-  pOperator->exprSupp.pExprInfo      = pExprInfo;
-  pOperator->exprSupp.numOfExprs = numOfCols;
   pOperator->info       = pInfo;
   pOperator->pTaskInfo  = pTaskInfo;
 
@@ -2360,9 +2417,6 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
     }
   }
   // semi interval operator does not catch result
-  if (!IS_FINAL_OP(pInfo)) {
-    pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
-  }
   pInfo->pUpdateRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
   pInfo->pUpdateRes->info.type = STREAM_REPROCESS;
   blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
@@ -2376,6 +2430,10 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
     pOperator->name = "StreamSemiIntervalOperator";
   }
 
+  if (!IS_FINAL_OP(pInfo)) {
+    pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
+  }
+
   pOperator->operatorType = pPhyNode->type;
   pOperator->blocking = true;
   pOperator->status = OP_NOT_OPENED;
@@ -2432,7 +2490,11 @@ void destroyStreamSessionAggOperatorInfo(void* param, int32_t numOfOutput) {
 }
 
 int32_t initBasicInfoEx(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock) {
-  initExprSupp(pSup, pExprInfo, numOfCols);
+  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
+
   initBasicInfo(pBasicInfo, pResultBlock);
   
   for (int32_t i = 0; i < numOfCols; ++i) {

source/libs/executor/src/tsort.c

@@ -374,11 +374,6 @@ int32_t msortComparFn(const void *pLeft, const void *pRight, void *param) {
   SSDataBlock* pLeftBlock = pLeftSource->src.pBlock;
   SSDataBlock* pRightBlock = pRightSource->src.pBlock;
 
-  // first sort by block groupId
-  if (pLeftBlock->info.groupId != pRightBlock->info.groupId) {
-    return pLeftBlock->info.groupId < pRightBlock->info.groupId ? -1 : 1;
-  }
-
   for(int32_t i = 0; i < pInfo->size; ++i) {
     SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, i);
     SColumnInfoData* pLeftColInfoData = TARRAY_GET_ELEM(pLeftBlock->pDataBlock, pOrder->slotId);

source/libs/nodes/src/nodesUtilFuncs.c

@@ -207,9 +207,12 @@ SNode* nodesMakeNode(ENodeType type) {
     case QUERY_NODE_SHOW_VNODES_STMT:
     case QUERY_NODE_SHOW_APPS_STMT:
     case QUERY_NODE_SHOW_SCORES_STMT:
-    case QUERY_NODE_SHOW_VARIABLE_STMT:
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
+    case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
     case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
       return makeNode(type, sizeof(SShowStmt));
+    case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
+      return makeNode(type, sizeof(SShowDnodeVariablesStmt));
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
       return makeNode(type, sizeof(SShowCreateDatabaseStmt));
     case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
@@ -637,13 +640,16 @@ void nodesDestroyNode(SNode* pNode) {
     case QUERY_NODE_SHOW_VNODES_STMT:
     case QUERY_NODE_SHOW_APPS_STMT:
     case QUERY_NODE_SHOW_SCORES_STMT:
-    case QUERY_NODE_SHOW_VARIABLE_STMT:
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
+    case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
     case QUERY_NODE_SHOW_TRANSACTIONS_STMT: {
       SShowStmt* pStmt = (SShowStmt*)pNode;
       nodesDestroyNode(pStmt->pDbName);
       nodesDestroyNode(pStmt->pTbName);
       break;
     }
+    case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:  // no pointer field
+      break;
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
       taosMemoryFreeClear(((SShowCreateDatabaseStmt*)pNode)->pCfg);
       break;

source/libs/parser/inc/parAst.h

@@ -161,6 +161,7 @@ SNode* createShowStmtWithCond(SAstCreateContext* pCxt, ENodeType type, SNode* pD
 SNode* createShowCreateDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName);
 SNode* createShowCreateTableStmt(SAstCreateContext* pCxt, ENodeType type, SNode* pRealTable);
 SNode* createShowTableDistributedStmt(SAstCreateContext* pCxt, SNode* pRealTable);
+SNode* createShowDnodeVariablesStmt(SAstCreateContext* pCxt, SNode* pDnodeId);
 SNode* createCreateUserStmt(SAstCreateContext* pCxt, SToken* pUserName, const SToken* pPassword);
 SNode* createAlterUserStmt(SAstCreateContext* pCxt, SToken* pUserName, int8_t alterType, const SToken* pVal);
 SNode* createDropUserStmt(SAstCreateContext* pCxt, SToken* pUserName);

source/libs/parser/inc/parUtil.h

@@ -47,6 +47,8 @@ typedef struct SParseMetaCache {
   SHashObj* pUserAuth;     // key is SUserAuthInfo serialized string, element is bool indicating whether or not to pass
   SHashObj* pUdf;          // key is funcName, element is SFuncInfo*
   SHashObj* pTableIndex;   // key is tbFName, element is SArray<STableIndexInfo>*
+  SArray*   pDnodes;       // element is SEpSet
+  bool      dnodeRequired;
 } SParseMetaCache;
 
 int32_t generateSyntaxErrMsg(SMsgBuf* pBuf, int32_t errCode, ...);
@@ -77,6 +79,7 @@ int32_t reserveUserAuthInCache(int32_t acctId, const char* pUser, const char* pD
 int32_t reserveUserAuthInCacheExt(const char* pUser, const SName* pName, AUTH_TYPE type, SParseMetaCache* pMetaCache);
 int32_t reserveUdfInCache(const char* pFunc, SParseMetaCache* pMetaCache);
 int32_t reserveTableIndexInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
+int32_t reserveDnodeRequiredInCache(SParseMetaCache* pMetaCache);
 int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta);
 int32_t getDbVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo);
 int32_t getTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup);
@@ -87,6 +90,7 @@ int32_t getUserAuthFromCache(SParseMetaCache* pMetaCache, const char* pUser, con
                              bool* pPass);
 int32_t getUdfInfoFromCache(SParseMetaCache* pMetaCache, const char* pFunc, SFuncInfo* pInfo);
 int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName, SArray** pIndexes);
+int32_t getDnodeListFromCache(SParseMetaCache* pMetaCache, SArray** pDnodes);
 void    destoryParseMetaCache(SParseMetaCache* pMetaCache);
 
 #ifdef __cplusplus

source/libs/parser/inc/sql.y

@@ -373,7 +373,9 @@ cmd ::= SHOW CREATE STABLE full_table_name(A).
 cmd ::= SHOW QUERIES.                                                             { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_QUERIES_STMT); }
 cmd ::= SHOW SCORES.                                                              { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_SCORES_STMT); }
 cmd ::= SHOW TOPICS.                                                              { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_TOPICS_STMT); }
-cmd ::= SHOW VARIABLES.                                                           { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_VARIABLE_STMT); }
+cmd ::= SHOW VARIABLES.                                                           { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_VARIABLES_STMT); }
+cmd ::= SHOW LOCAL VARIABLES.                                                     { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT); }
+cmd ::= SHOW DNODE NK_INTEGER(A) VARIABLES.                                       { pCxt->pRootNode = createShowDnodeVariablesStmt(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &A)); }
 cmd ::= SHOW BNODES.                                                              { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_BNODES_STMT); }
 cmd ::= SHOW SNODES.                                                              { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_SNODES_STMT); }
 cmd ::= SHOW CLUSTER.                                                             { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_CLUSTER_STMT); }

source/libs/parser/src/parAstCreater.c

@@ -1197,6 +1197,14 @@ SNode* createShowTableDistributedStmt(SAstCreateContext* pCxt, SNode* pRealTable
   return (SNode*)pStmt;
 }
 
+SNode* createShowDnodeVariablesStmt(SAstCreateContext* pCxt, SNode* pDnodeId) {
+  CHECK_PARSER_STATUS(pCxt);
+  SShowDnodeVariablesStmt* pStmt = (SShowDnodeVariablesStmt*)nodesMakeNode(QUERY_NODE_SHOW_DNODE_VARIABLES_STMT);
+  CHECK_OUT_OF_MEM(pStmt);
+  pStmt->pDnodeId = pDnodeId;
+  return (SNode*)pStmt;
+}
+
 SNode* createCreateUserStmt(SAstCreateContext* pCxt, SToken* pUserName, const SToken* pPassword) {
   CHECK_PARSER_STATUS(pCxt);
   char password[TSDB_USET_PASSWORD_LEN] = {0};

source/libs/parser/src/parAstParser.c

@@ -397,6 +397,15 @@ static int32_t collectMetaKeyFromShowVariables(SCollectMetaKeyCxt* pCxt, SShowSt
                                  pCxt->pMetaCache);
 }
 
+static int32_t collectMetaKeyFromShowDnodeVariables(SCollectMetaKeyCxt* pCxt, SShowDnodeVariablesStmt* pStmt) {
+  int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB,
+                                         TSDB_INS_TABLE_DNODE_VARIABLES, pCxt->pMetaCache);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = reserveDnodeRequiredInCache(pCxt->pMetaCache);
+  }
+  return code;
+}
+
 static int32_t collectMetaKeyFromShowCreateDatabase(SCollectMetaKeyCxt* pCxt, SShowCreateDatabaseStmt* pStmt) {
   return reserveDbCfgInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pCxt->pMetaCache);
 }
@@ -502,8 +511,10 @@ static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
       return collectMetaKeyFromShowConnections(pCxt, (SShowStmt*)pStmt);
     case QUERY_NODE_SHOW_QUERIES_STMT:
       return collectMetaKeyFromShowQueries(pCxt, (SShowStmt*)pStmt);
-    case QUERY_NODE_SHOW_VARIABLE_STMT:
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
       return collectMetaKeyFromShowVariables(pCxt, (SShowStmt*)pStmt);
+    case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
+      return collectMetaKeyFromShowDnodeVariables(pCxt, (SShowDnodeVariablesStmt*)pStmt);
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
       return collectMetaKeyFromShowCreateDatabase(pCxt, (SShowCreateDatabaseStmt*)pStmt);
     case QUERY_NODE_SHOW_CREATE_TABLE_STMT:

source/libs/parser/src/parTranslater.c

@@ -305,6 +305,24 @@ static int32_t getTableIndex(STranslateContext* pCxt, const SName* pName, SArray
   return code;
 }
 
+static int32_t getDnodeList(STranslateContext* pCxt, SArray** pDnodes) {
+  SParseContext* pParCxt = pCxt->pParseCxt;
+  int32_t        code = TSDB_CODE_SUCCESS;
+  if (pParCxt->async) {
+    code = getDnodeListFromCache(pCxt->pMetaCache, pDnodes);
+  } else {
+    SRequestConnInfo conn = {.pTrans = pParCxt->pTransporter,
+                             .requestId = pParCxt->requestId,
+                             .requestObjRefId = pParCxt->requestRid,
+                             .mgmtEps = pParCxt->mgmtEpSet};
+    code = catalogGetDnodeList(pParCxt->pCatalog, &conn, pDnodes);
+  }
+  if (TSDB_CODE_SUCCESS != code) {
+    parserError("getDnodeList error, code:%s", tstrerror(code));
+  }
+  return code;
+}
+
 static int32_t initTranslateContext(SParseContext* pParseCxt, SParseMetaCache* pMetaCache, STranslateContext* pCxt) {
   pCxt->pParseCxt = pParseCxt;
   pCxt->errCode = TSDB_CODE_SUCCESS;
@@ -1450,12 +1468,27 @@ static int32_t addMnodeToVgroupList(const SEpSet* pEpSet, SArray** pVgroupList)
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t setSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRealTableNode* pRealTable) {
+static int32_t dnodeToVgroupsInfo(SArray* pDnodes, SVgroupsInfo** pVgsInfo) {
-  if (0 != strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_USER_TABLES) &&
+  size_t ndnode = taosArrayGetSize(pDnodes);
-      0 != strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED)) {
+  *pVgsInfo = taosMemoryCalloc(1, sizeof(SVgroupsInfo) + sizeof(SVgroupInfo) * ndnode);
-    return TSDB_CODE_SUCCESS;
+  if (NULL == *pVgsInfo) {
+    return TSDB_CODE_OUT_OF_MEMORY;
   }
+  (*pVgsInfo)->numOfVgroups = ndnode;
+  for (int32_t i = 0; i < ndnode; ++i) {
+    memcpy(&((*pVgsInfo)->vgroups[i].epSet), taosArrayGet(pDnodes, i), sizeof(SEpSet));
+  }
+  return TSDB_CODE_SUCCESS;
+}
 
+static bool sysTableFromVnode(const char* pTable) {
+  return (0 == strcmp(pTable, TSDB_INS_TABLE_USER_TABLES)) ||
+         (0 == strcmp(pTable, TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED));
+}
+
+static bool sysTableFromDnode(const char* pTable) { return 0 == strcmp(pTable, TSDB_INS_TABLE_DNODE_VARIABLES); }
+
+static int32_t setVnodeSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRealTableNode* pRealTable) {
   int32_t code = TSDB_CODE_SUCCESS;
   SArray* vgroupList = NULL;
   if ('\0' != pRealTable->qualDbName[0]) {
@@ -1478,6 +1511,26 @@ static int32_t setSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRea
   return code;
 }
 
+static int32_t setDnodeSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRealTableNode* pRealTable) {
+  SArray* pDnodes = NULL;
+  int32_t code = getDnodeList(pCxt, &pDnodes);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = dnodeToVgroupsInfo(pDnodes, &pRealTable->pVgroupList);
+  }
+  taosArrayDestroy(pDnodes);
+  return code;
+}
+
+static int32_t setSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRealTableNode* pRealTable) {
+  if (sysTableFromVnode(pRealTable->table.tableName)) {
+    return setVnodeSysTableVgroupList(pCxt, pName, pRealTable);
+  } else if (sysTableFromDnode(pRealTable->table.tableName)) {
+    return setDnodeSysTableVgroupList(pCxt, pName, pRealTable);
+  } else {
+    return TSDB_CODE_SUCCESS;
+  }
+}
+
 static int32_t setTableVgroupList(STranslateContext* pCxt, SName* pName, SRealTableNode* pRealTable) {
   if (pCxt->pParseCxt->topicQuery) {
     return TSDB_CODE_SUCCESS;
@@ -4081,8 +4134,12 @@ static int32_t translateCreateStream(STranslateContext* pCxt, SCreateStreamStmt*
 }
 
 static int32_t translateDropStream(STranslateContext* pCxt, SDropStreamStmt* pStmt) {
-  // todo
+  SMDropStreamReq dropReq = {0};
-  return TSDB_CODE_SUCCESS;
+  SName           name;
+  tNameSetDbName(&name, pCxt->pParseCxt->acctId, pStmt->streamName, strlen(pStmt->streamName));
+  tNameGetFullDbName(&name, dropReq.name);
+  dropReq.igNotExists = pStmt->ignoreNotExists;
+  return buildCmdMsg(pCxt, TDMT_MND_DROP_STREAM, (FSerializeFunc)tSerializeSMDropStreamReq, &dropReq);
 }
 
 static int32_t readFromFile(char* pName, int32_t* len, char** buf) {
@@ -4538,7 +4595,8 @@ static const char* getSysDbName(ENodeType type) {
     case QUERY_NODE_SHOW_SNODES_STMT:
     case QUERY_NODE_SHOW_LICENCE_STMT:
     case QUERY_NODE_SHOW_CLUSTER_STMT:
-    case QUERY_NODE_SHOW_VARIABLE_STMT:
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
+    case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
       return TSDB_INFORMATION_SCHEMA_DB;
     case QUERY_NODE_SHOW_CONNECTIONS_STMT:
     case QUERY_NODE_SHOW_QUERIES_STMT:
@@ -4595,10 +4653,12 @@ static const char* getSysTableName(ENodeType type) {
       return TSDB_PERFS_TABLE_TOPICS;
     case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
       return TSDB_PERFS_TABLE_TRANS;
-    case QUERY_NODE_SHOW_VARIABLE_STMT:
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
       return TSDB_INS_TABLE_CONFIGS;
     case QUERY_NODE_SHOW_APPS_STMT:
       return TSDB_PERFS_TABLE_APPS;
+    case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
+      return TSDB_INS_TABLE_DNODE_VARIABLES;
     default:
       break;
   }
@@ -4725,6 +4785,21 @@ static int32_t rewriteShow(STranslateContext* pCxt, SQuery* pQuery) {
   return code;
 }
 
+static int32_t rewriteShowDnodeVariables(STranslateContext* pCxt, SQuery* pQuery) {
+  SSelectStmt* pStmt = NULL;
+  int32_t      code = createSelectStmtForShow(nodeType(pQuery->pRoot), &pStmt);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = createOperatorNode(OP_TYPE_EQUAL, "dnode_id", ((SShowDnodeVariablesStmt*)pQuery->pRoot)->pDnodeId,
+                              &pStmt->pWhere);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    pQuery->showRewrite = true;
+    nodesDestroyNode(pQuery->pRoot);
+    pQuery->pRoot = (SNode*)pStmt;
+  }
+  return code;
+}
+
 static SNode* createBlockDistInfoFunc() {
   SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
   if (NULL == pFunc) {
@@ -5720,10 +5795,13 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
     case QUERY_NODE_SHOW_CLUSTER_STMT:
     case QUERY_NODE_SHOW_TOPICS_STMT:
     case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
-    case QUERY_NODE_SHOW_VARIABLE_STMT:
+    case QUERY_NODE_SHOW_VARIABLES_STMT:
     case QUERY_NODE_SHOW_APPS_STMT:
       code = rewriteShow(pCxt, pQuery);
       break;
+    case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
+      code = rewriteShowDnodeVariables(pCxt, pQuery);
+      break;
     case QUERY_NODE_SHOW_TABLE_DISTRIBUTED_STMT:
       code = rewriteShowTableDist(pCxt, pQuery);
       break;
@@ -5814,6 +5892,7 @@ static int32_t setQuery(STranslateContext* pCxt, SQuery* pQuery) {
       break;
     case QUERY_NODE_RESET_QUERY_CACHE_STMT:
     case QUERY_NODE_ALTER_LOCAL_STMT:
+    case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
       pQuery->execMode = QUERY_EXEC_MODE_LOCAL;
       break;
     default:

source/libs/parser/src/parUtil.c

@@ -561,6 +561,7 @@ int32_t buildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalog
   if (TSDB_CODE_SUCCESS == code) {
     code = buildTableReq(pMetaCache->pTableIndex, &pCatalogReq->pTableIndex);
   }
+  pCatalogReq->dNodeRequired = pMetaCache->dnodeRequired;
   return code;
 }
 
@@ -656,6 +657,7 @@ int32_t putMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMet
   if (TSDB_CODE_SUCCESS == code) {
     code = putTableDataToCache(pCatalogReq->pTableIndex, pMetaData->pTableIndex, &pMetaCache->pTableIndex);
   }
+  pMetaCache->pDnodes = pMetaData->pDnodeList;
   return code;
 }
 
@@ -875,6 +877,19 @@ int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName,
   return code;
 }
 
+int32_t reserveDnodeRequiredInCache(SParseMetaCache* pMetaCache) {
+  pMetaCache->dnodeRequired = true;
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t getDnodeListFromCache(SParseMetaCache* pMetaCache, SArray** pDnodes) {
+  *pDnodes = taosArrayDup(pMetaCache->pDnodes);
+  if (NULL == *pDnodes) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 void destoryParseMetaCache(SParseMetaCache* pMetaCache) {
   taosHashCleanup(pMetaCache->pTableMeta);
   taosHashCleanup(pMetaCache->pDbVgroup);

source/libs/parser/test/mockCatalog.cpp

@@ -93,6 +93,16 @@ void generateInformationSchema(MockCatalogService* mcs) {
                                  .addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN);
     builder.done();
   }
+  {
+    ITableBuilder& builder = mcs->createTableBuilder("information_schema", "configs", TSDB_SYSTEM_TABLE, 1)
+                                 .addColumn("name", TSDB_DATA_TYPE_BINARY, TSDB_CONFIG_OPTION_LEN);
+    builder.done();
+  }
+  {
+    ITableBuilder& builder = mcs->createTableBuilder("information_schema", "dnode_variables", TSDB_SYSTEM_TABLE, 1)
+                                 .addColumn("dnode_id", TSDB_DATA_TYPE_INT);
+    builder.done();
+  }
 }
 
 void generatePerformanceSchema(MockCatalogService* mcs) {
@@ -187,6 +197,12 @@ void generateFunctions(MockCatalogService* mcs) {
                       8);
 }
 
+void generateDnodes(MockCatalogService* mcs) {
+  mcs->createDnode(1, "host1", 7030);
+  mcs->createDnode(2, "host2", 7030);
+  mcs->createDnode(3, "host3", 7030);
+}
+
 }  // namespace
 
 int32_t __catalogGetHandle(const char* clusterId, struct SCatalog** catalogHandle) { return 0; }
@@ -241,6 +257,10 @@ int32_t __catalogGetTableIndex(SCatalog* pCtg, void* pTrans, const SEpSet* pMgmt
   return g_mockCatalogService->catalogGetTableIndex(pName, pRes);
 }
 
+int32_t __catalogGetDnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray** pDnodeList) {
+  return g_mockCatalogService->catalogGetDnodeList(pDnodeList);
+}
+
 void initMetaDataEnv() {
   g_mockCatalogService.reset(new MockCatalogService());
 
@@ -258,6 +278,7 @@ void initMetaDataEnv() {
   stub.set(catalogRefreshGetTableMeta, __catalogRefreshGetTableMeta);
   stub.set(catalogRemoveTableMeta, __catalogRemoveTableMeta);
   stub.set(catalogGetTableIndex, __catalogGetTableIndex);
+  stub.set(catalogGetDnodeList, __catalogGetDnodeList);
   // {
   //   AddrAny any("libcatalog.so");
   //   std::map<std::string,void*> result;
@@ -307,6 +328,7 @@ void generateMetaData() {
   generateTestST1(g_mockCatalogService.get());
   generateTestST2(g_mockCatalogService.get());
   generateFunctions(g_mockCatalogService.get());
+  generateDnodes(g_mockCatalogService.get());
   g_mockCatalogService->showTables();
 }
 

source/libs/parser/test/mockCatalogService.cpp

@@ -165,6 +165,14 @@ class MockCatalogServiceImpl {
     return TSDB_CODE_SUCCESS;
   }
 
+  int32_t catalogGetDnodeList(SArray** pDnodes) const {
+    *pDnodes = taosArrayInit(dnode_.size(), sizeof(SEpSet));
+    for (const auto& dnode : dnode_) {
+      taosArrayPush(*pDnodes, &dnode.second);
+    }
+    return TSDB_CODE_SUCCESS;
+  }
+
   int32_t catalogGetAllMeta(const SCatalogReq* pCatalogReq, SMetaData* pMetaData) const {
     int32_t code = getAllTableMeta(pCatalogReq->pTableMeta, &pMetaData->pTableMeta);
     if (TSDB_CODE_SUCCESS == code) {
@@ -188,6 +196,9 @@ class MockCatalogServiceImpl {
     if (TSDB_CODE_SUCCESS == code) {
       code = getAllTableIndex(pCatalogReq->pTableIndex, &pMetaData->pTableIndex);
     }
+    if (TSDB_CODE_SUCCESS == code && pCatalogReq->dNodeRequired) {
+      code = catalogGetDnodeList(&pMetaData->pDnodeList);
+    }
     return code;
   }
 
@@ -303,11 +314,18 @@ class MockCatalogServiceImpl {
     }
   }
 
+  void createDnode(int32_t dnodeId, const std::string& host, int16_t port) {
+    SEpSet epSet = {0};
+    addEpIntoEpSet(&epSet, host.c_str(), port);
+    dnode_.insert(std::make_pair(dnodeId, epSet));
+  }
+
  private:
   typedef std::map<std::string, std::shared_ptr<MockTableMeta>> TableMetaCache;
   typedef std::map<std::string, TableMetaCache>                 DbMetaCache;
   typedef std::map<std::string, std::shared_ptr<SFuncInfo>>     UdfMetaCache;
   typedef std::map<std::string, std::vector<STableIndexInfo>>   IndexMetaCache;
+  typedef std::map<int32_t, SEpSet>                             DnodeCache;
 
   uint64_t getNextId() { return id_++; }
 
@@ -532,6 +550,7 @@ class MockCatalogServiceImpl {
   DbMetaCache                   meta_;
   UdfMetaCache                  udf_;
   IndexMetaCache                index_;
+  DnodeCache                    dnode_;
 };
 
 MockCatalogService::MockCatalogService() : impl_(new MockCatalogServiceImpl()) {}
@@ -557,6 +576,10 @@ void MockCatalogService::createFunction(const std::string& func, int8_t funcType
 
 void MockCatalogService::createSmaIndex(const SMCreateSmaReq* pReq) { impl_->createSmaIndex(pReq); }
 
+void MockCatalogService::createDnode(int32_t dnodeId, const std::string& host, int16_t port) {
+  impl_->createDnode(dnodeId, host, port);
+}
+
 int32_t MockCatalogService::catalogGetTableMeta(const SName* pTableName, STableMeta** pTableMeta) const {
   return impl_->catalogGetTableMeta(pTableName, pTableMeta);
 }
@@ -581,6 +604,8 @@ int32_t MockCatalogService::catalogGetTableIndex(const SName* pTableName, SArray
   return impl_->catalogGetTableIndex(pTableName, pIndexes);
 }
 
+int32_t MockCatalogService::catalogGetDnodeList(SArray** pDnodes) const { return impl_->catalogGetDnodeList(pDnodes); }
+
 int32_t MockCatalogService::catalogGetAllMeta(const SCatalogReq* pCatalogReq, SMetaData* pMetaData) const {
   return impl_->catalogGetAllMeta(pCatalogReq, pMetaData);
 }

source/libs/parser/test/mockCatalogService.h

@@ -62,6 +62,7 @@ class MockCatalogService {
   void showTables() const;
   void createFunction(const std::string& func, int8_t funcType, int8_t outputType, int32_t outputLen, int32_t bufSize);
   void createSmaIndex(const SMCreateSmaReq* pReq);
+  void createDnode(int32_t dnodeId, const std::string& host, int16_t port);
 
   int32_t catalogGetTableMeta(const SName* pTableName, STableMeta** pTableMeta) const;
   int32_t catalogGetTableHashVgroup(const SName* pTableName, SVgroupInfo* vgInfo) const;
@@ -69,6 +70,7 @@ class MockCatalogService {
   int32_t catalogGetDBVgInfo(const char* pDbFName, SArray** pVgList) const;
   int32_t catalogGetUdfInfo(const std::string& funcName, SFuncInfo* pInfo) const;
   int32_t catalogGetTableIndex(const SName* pTableName, SArray** pIndexes) const;
+  int32_t catalogGetDnodeList(SArray** pDnodes) const;
   int32_t catalogGetAllMeta(const SCatalogReq* pCatalogReq, SMetaData* pMetaData) const;
 
  private:

source/libs/parser/test/parInitialDTest.cpp

@@ -159,7 +159,35 @@ TEST_F(ParserInitialDTest, dropSTable) {
   run("DROP STABLE st1");
 }
 
-// todo DROP stream
+TEST_F(ParserInitialDTest, dropStream) {
+  useDb("root", "test");
+
+  SMDropStreamReq expect = {0};
+
+  auto clearDropStreamReq = [&]() { memset(&expect, 0, sizeof(SMDropStreamReq)); };
+
+  auto setDropStreamReq = [&](const char* pStream, int8_t igNotExists = 0) {
+    sprintf(expect.name, "0.%s", pStream);
+    expect.igNotExists = igNotExists;
+  };
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_DROP_STREAM_STMT);
+    SMDropStreamReq req = {0};
+    ASSERT_TRUE(TSDB_CODE_SUCCESS == tDeserializeSMDropStreamReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req));
+
+    ASSERT_EQ(std::string(req.name), std::string(expect.name));
+    ASSERT_EQ(req.igNotExists, expect.igNotExists);
+  });
+
+  setDropStreamReq("s1");
+  run("DROP STREAM s1");
+  clearDropStreamReq();
+
+  setDropStreamReq("s2", 1);
+  run("DROP STREAM IF EXISTS s2");
+  clearDropStreamReq();
+}
 
 TEST_F(ParserInitialDTest, dropTable) {
   useDb("root", "test");

source/libs/parser/test/parShowToUse.cpp

@@ -76,6 +76,12 @@ TEST_F(ParserShowToUseTest, showDnodes) {
   run("SHOW dnodes");
 }
 
+TEST_F(ParserShowToUseTest, showDnodeVariables) {
+  useDb("root", "test");
+
+  run("SHOW DNODE 1 VARIABLES");
+}
+
 TEST_F(ParserShowToUseTest, showFunctions) {
   useDb("root", "test");
 
@@ -84,6 +90,12 @@ TEST_F(ParserShowToUseTest, showFunctions) {
 
 // todo SHOW licence
 
+TEST_F(ParserShowToUseTest, showLocalVariables) {
+  useDb("root", "test");
+
+  run("SHOW LOCAL VARIABLES");
+}
+
 TEST_F(ParserShowToUseTest, showIndexes) {
   useDb("root", "test");
 
@@ -157,7 +169,11 @@ TEST_F(ParserShowToUseTest, showUsers) {
   run("SHOW users");
 }
 
-// todo SHOW variables
+TEST_F(ParserShowToUseTest, showVariables) {
+  useDb("root", "test");
+
+  run("SHOW VARIABLES");
+}
 
 TEST_F(ParserShowToUseTest, showVgroups) {
   useDb("root", "test");

source/libs/planner/src/planOptimizer.c

@@ -1056,49 +1056,69 @@ static bool partTagsOptHasCol(SNodeList* pPartKeys) {
   return hasCol;
 }
 
+static bool partTagsIsOptimizableNode(SLogicNode* pNode) {
+  return ((QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode) /*||
+           (QUERY_NODE_LOGIC_PLAN_AGG == nodeType(pNode) && NULL != ((SAggLogicNode*)pNode)->pGroupKeys &&
+            NULL != ((SAggLogicNode*)pNode)->pAggFuncs)*/) &&
+          1 == LIST_LENGTH(pNode->pChildren) &&
+          QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(nodesListGetNode(pNode->pChildren, 0)));
+}
+
+static SNodeList* partTagsGetPartKeys(SLogicNode* pNode) {
+  if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
+    return ((SPartitionLogicNode*)pNode)->pPartitionKeys;
+  } else {
+    return ((SAggLogicNode*)pNode)->pGroupKeys;
+  }
+}
+
 static bool partTagsOptMayBeOptimized(SLogicNode* pNode) {
-  if (QUERY_NODE_LOGIC_PLAN_PARTITION != nodeType(pNode) || 1 != LIST_LENGTH(pNode->pChildren) ||
+  if (!partTagsIsOptimizableNode(pNode)) {
-      QUERY_NODE_LOGIC_PLAN_SCAN != nodeType(nodesListGetNode(pNode->pChildren, 0))) {
     return false;
   }
 
-  return !partTagsOptHasCol(((SPartitionLogicNode*)pNode)->pPartitionKeys);
+  return !partTagsOptHasCol(partTagsGetPartKeys(pNode));
 }
 
 static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
-  SPartitionLogicNode* pPart =
+  SLogicNode* pNode = optFindPossibleNode(pLogicSubplan->pNode, partTagsOptMayBeOptimized);
-      (SPartitionLogicNode*)optFindPossibleNode(pLogicSubplan->pNode, partTagsOptMayBeOptimized);
+  if (NULL == pNode) {
-  if (NULL == pPart) {
     return TSDB_CODE_SUCCESS;
   }
 
-  SScanLogicNode* pScan = (SScanLogicNode*)nodesListGetNode(pPart->node.pChildren, 0);
+  int32_t         code = TSDB_CODE_SUCCESS;
-  TSWAP(pPart->pPartitionKeys, pScan->pPartTags);
+  SScanLogicNode* pScan = (SScanLogicNode*)nodesListGetNode(pNode->pChildren, 0);
-  int32_t code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pPart, (SLogicNode*)pScan);
+  if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
-  if (TSDB_CODE_SUCCESS == code) {
+    TSWAP(((SPartitionLogicNode*)pNode)->pPartitionKeys, pScan->pPartTags);
-    NODES_CLEAR_LIST(pPart->node.pChildren);
+    int32_t code = replaceLogicNode(pLogicSubplan, pNode, (SLogicNode*)pScan);
-    nodesDestroyNode((SNode*)pPart);
+    if (TSDB_CODE_SUCCESS == code) {
+      NODES_CLEAR_LIST(pNode->pChildren);
+      nodesDestroyNode((SNode*)pNode);
+    }
+  } else {
+    TSWAP(((SAggLogicNode*)pNode)->pGroupKeys, pScan->pPartTags);
   }
   return code;
 }
 
 static bool eliminateProjOptMayBeOptimized(SLogicNode* pNode) {
-  //TODO: enable this optimization after new mechanising that map projection and targets of project node
+  // TODO: enable this optimization after new mechanising that map projection and targets of project node
   if (NULL != pNode->pParent) {
     return false;
   }
-  
+
   if (QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(pNode) || 1 != LIST_LENGTH(pNode->pChildren)) {
     return false;
   }
 
   SProjectLogicNode* pProjectNode = (SProjectLogicNode*)pNode;
-  if (-1 != pProjectNode->limit || -1 != pProjectNode->slimit || -1 != pProjectNode->offset  || -1 != pProjectNode->soffset) {
+  if (-1 != pProjectNode->limit || -1 != pProjectNode->slimit || -1 != pProjectNode->offset ||
+      -1 != pProjectNode->soffset) {
     return false;
   }
 
   SHashObj* pProjColNameHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
-  SNode* pProjection;
+  SNode*    pProjection;
   FOREACH(pProjection, pProjectNode->pProjections) {
     SExprNode* pExprNode = (SExprNode*)pProjection;
     if (QUERY_NODE_COLUMN != nodeType(pExprNode)) {
@@ -1106,7 +1126,7 @@ static bool eliminateProjOptMayBeOptimized(SLogicNode* pNode) {
       return false;
     }
 
-    char* projColumnName = ((SColumnNode*)pProjection)->colName;
+    char*    projColumnName = ((SColumnNode*)pProjection)->colName;
     int32_t* pExist = taosHashGet(pProjColNameHash, projColumnName, strlen(projColumnName));
     if (NULL != pExist) {
       taosHashCleanup(pProjColNameHash);
@@ -1121,9 +1141,10 @@ static bool eliminateProjOptMayBeOptimized(SLogicNode* pNode) {
   return true;
 }
 
-static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan, SProjectLogicNode* pProjectNode) {
+static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan,
+                                         SProjectLogicNode* pProjectNode) {
   SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pProjectNode->node.pChildren, 0);
-  SNodeList* pNewChildTargets = nodesMakeList();
+  SNodeList*  pNewChildTargets = nodesMakeList();
 
   SNode* pProjection = NULL;
   FOREACH(pProjection, pProjectNode->pProjections) {
@@ -1137,7 +1158,7 @@ static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan*
   }
   nodesDestroyList(pChild->pTargets);
   pChild->pTargets = pNewChildTargets;
-  
+
   int32_t code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pProjectNode, pChild);
   if (TSDB_CODE_SUCCESS == code) {
     NODES_CLEAR_LIST(pProjectNode->node.pChildren);
@@ -1148,7 +1169,7 @@ static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan*
 
 static int32_t eliminateProjOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
   SProjectLogicNode* pProjectNode =
-    (SProjectLogicNode*)optFindPossibleNode(pLogicSubplan->pNode, eliminateProjOptMayBeOptimized);
+      (SProjectLogicNode*)optFindPossibleNode(pLogicSubplan->pNode, eliminateProjOptMayBeOptimized);
 
   if (NULL == pProjectNode) {
     return TSDB_CODE_SUCCESS;

source/libs/planner/src/planPhysiCreater.c

@@ -534,7 +534,11 @@ static int32_t createSystemTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan*
   }
   SQueryNodeLoad node = {.addr = {.nodeId = MNODE_HANDLE, .epSet = pCxt->pPlanCxt->mgmtEpSet}, .load = 0};
   taosArrayPush(pCxt->pExecNodeList, &node);
-  pScan->mgmtEpSet = pCxt->pPlanCxt->mgmtEpSet;
+  if (0 == strcmp(pScanLogicNode->tableName.tname, TSDB_INS_TABLE_DNODE_VARIABLES)) {
+    pScan->mgmtEpSet = pScanLogicNode->pVgroupList->vgroups->epSet;
+  } else {
+    pScan->mgmtEpSet = pCxt->pPlanCxt->mgmtEpSet;
+  }
   tNameGetFullDbName(&pScanLogicNode->tableName, pSubplan->dbFName);
 
   return createScanPhysiNodeFinalize(pCxt, pSubplan, pScanLogicNode, (SScanPhysiNode*)pScan, pPhyNode);
@@ -879,12 +883,15 @@ static int32_t createInterpFuncPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pCh
     pInterpFunc->interval = pFuncLogicNode->interval;
     pInterpFunc->fillMode = pFuncLogicNode->fillMode;
     pInterpFunc->pFillValues = nodesCloneNode(pFuncLogicNode->pFillValues);
-    pInterpFunc->pTimeSeries = nodesCloneNode(pFuncLogicNode->pTimeSeries);
+    if (NULL != pFuncLogicNode->pFillValues && NULL == pInterpFunc->pFillValues) {
-    if (NULL == pInterpFunc->pTimeSeries || (NULL != pFuncLogicNode->pFillValues && NULL == pInterpFunc->pFillValues)) {
       code = TSDB_CODE_OUT_OF_MEMORY;
     }
   }
 
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setNodeSlotId(pCxt, pChildTupe->dataBlockId, -1, pFuncLogicNode->pTimeSeries, &pInterpFunc->pTimeSeries);
+  }
+
   if (TSDB_CODE_SUCCESS == code) {
     *pPhyNode = (SPhysiNode*)pInterpFunc;
   } else {

source/libs/planner/test/planOptimizeTest.cpp

@@ -45,14 +45,21 @@ TEST_F(PlanOptimizeTest, ConditionPushDown) {
 TEST_F(PlanOptimizeTest, orderByPrimaryKey) {
   useDb("root", "test");
 
-  run("SELECT * FROM t1 ORDER BY ts");
-  run("SELECT * FROM t1 ORDER BY ts DESC");
   run("SELECT c1 FROM t1 ORDER BY ts");
+
   run("SELECT c1 FROM t1 ORDER BY ts DESC");
 
   run("SELECT COUNT(*) FROM t1 INTERVAL(10S) ORDER BY _WSTARTTS DESC");
 }
 
+TEST_F(PlanOptimizeTest, PartitionTags) {
+  useDb("root", "test");
+
+  run("SELECT c1 FROM st1 PARTITION BY tag1");
+
+  run("SELECT SUM(c1) FROM st1 GROUP BY tag1");
+}
+
 TEST_F(PlanOptimizeTest, eliminateProjection) {
   useDb("root", "test");
 
@@ -60,5 +67,5 @@ TEST_F(PlanOptimizeTest, eliminateProjection) {
   run("SELECT c1 FROM t1");
   run("SELECT * FROM st1");
   run("SELECT c1 FROM st1s3");
-  //run("select 1-abs(c1) from (select unique(c1) c1 from st1s3) order by 1 nulls first");
+  // run("select 1-abs(c1) from (select unique(c1) c1 from st1s3) order by 1 nulls first");
 }

source/libs/planner/test/planOtherTest.cpp

@@ -74,6 +74,8 @@ TEST_F(PlanOtherTest, show) {
   run("SHOW TABLE DISTRIBUTED t1");
 
   run("SHOW TABLE DISTRIBUTED st1");
+
+  run("SHOW DNODE 1 VARIABLES");
 }
 
 TEST_F(PlanOtherTest, delete) {

source/libs/sync/src/syncEnv.c

@@ -35,8 +35,9 @@ bool syncEnvIsStart() {
 }
 
 int32_t syncEnvStart() {
-  int32_t ret = 0;
+  int32_t  ret = 0;
-  taosSeedRand(taosGetTimestampSec());
+  uint32_t seed = (uint32_t)(taosGetTimestampNs() & 0x00000000FFFFFFFF);
+  taosSeedRand(seed);
   // gSyncEnv = doSyncEnvStart(gSyncEnv);
   gSyncEnv = doSyncEnvStart();
   assert(gSyncEnv != NULL);

source/libs/sync/src/syncMain.c

@@ -154,10 +154,14 @@ int32_t syncSetStandby(int64_t rid) {
     return -1;
   }
 
-  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
+  if (pSyncNode->state != TAOS_SYNC_STATE_FOLLOWER) {
     taosReleaseRef(tsNodeRefId, pSyncNode->rid);
-    terrno = TSDB_CODE_SYN_IS_LEADER;
+    if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
-    sError("failed to set standby since it is leader, rid:%" PRId64, rid);
+      terrno = TSDB_CODE_SYN_IS_LEADER;
+    } else {
+      terrno = TSDB_CODE_SYN_STANDBY_NOT_READY;
+    }
+    sError("failed to set standby since it is not follower, state:%s rid:%" PRId64, syncStr(pSyncNode->state), rid);
     return -1;
   }
 
@@ -2290,7 +2294,7 @@ static int32_t syncNodeConfigChangeFinish(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyn
     ths->pFsm->FpReConfigCb(ths->pFsm, pRpcMsg, cbMeta);
   }
 
-  // update changing
+  // clear changing
   ths->changing = false;
 
   char  tmpbuf[512];
@@ -2309,6 +2313,9 @@ static int32_t syncNodeConfigChangeFinish(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyn
 
 static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry,
                                     SyncReconfigFinish* pFinish) {
+  // set changing
+  ths->changing = true;
+
   // old config
   SSyncCfg oldSyncCfg = ths->pRaftCfg->cfg;
 

source/util/src/terror.c

@@ -426,6 +426,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_IN_NEW_CONFIG,        "Sync not in new confi
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NEW_CONFIG_ERROR,         "Sync new config error")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_RECONFIG_NOT_READY,       "Sync not ready for reconfig")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_PROPOSE_NOT_READY,        "Sync not ready for propose")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_STANDBY_NOT_READY,        "Sync not ready for standby")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR,           "Sync internal error")
 
 // wal

tests/script/jenkins/basic.txt

@@ -81,6 +81,7 @@
 # ./test.sh -f tsim/stream/triggerSession0.sim
 ./test.sh -f tsim/stream/partitionby.sim
 ./test.sh -f tsim/stream/schedSnode.sim
+./test.sh -f tsim/stream/windowClose.sim
 
 
 # ---- transaction

tests/script/tsim/mnode/basic5.sim

@@ -148,7 +148,7 @@ endi
 
 print =============== step6: stop mnode1 
 system sh/exec.sh -n dnode1 -s stop -x SIGKILL
-sql_error drop mnode on dnode 1
+# sql_error drop mnode on dnode 1
 
 $x = 0
 step61: 

tests/script/tsim/stream/windowClose.sim

@@ -0,0 +1,32 @@
+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
+
+print =============== create database
+sql create database test vgroups 1
+sql show databases
+if $rows != 3 then
+  return -1
+endi
+
+print $data00 $data01 $data02
+
+sql use test
+sql create stable st(ts timestamp, a int) tags(t int);
+sql create table tu1 using st tags(1);
+sql create table tu2 using st tags(2);
+
+sql create stream stream1 trigger window_close into streamt as select  _wstartts, sum(a) from st interval(10s);
+
+sql insert into tu1 values(now, 1);
+
+sleep 500
+sql select * from streamt;
+if $rows != 0 then
+  print ======$rows
+  return -1
+endi
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/system-test/7-tmq/subscribeDb3.py

@@ -54,9 +54,11 @@ class TDTestCase:
         tdSql.query("create database if not exists %s vgroups 1"%(cdbName))
         tdSql.query("drop table if exists %s.consumeinfo "%(cdbName))
         tdSql.query("drop table if exists %s.consumeresult "%(cdbName))
+        tdSql.query("drop table if exists %s.notifyinfo "%(cdbName))        
 
         tdSql.query("create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)"%cdbName)
         tdSql.query("create table %s.consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"%cdbName)
+        tdSql.query("create table %s.notifyinfo (ts timestamp, cmdid int, consumerid int)"%cdbName)
 
     def insertConsumerInfo(self,consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifmanualcommit,cdbName='cdb'):    
         sql = "insert into %s.consumeinfo values "%cdbName
@@ -64,6 +66,27 @@ class TDTestCase:
         tdLog.info("consume info sql: %s"%sql)
         tdSql.query(sql)
 
+    def getStartConsumeNotifyFromTmqsim(self,cdbName='cdb'):
+        while 1:
+            tdSql.query("select * from %s.notifyinfo"%cdbName)
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            if (tdSql.getRows() == 1) and (tdSql.getData(0, 1) == 0):
+                break
+            else:
+                time.sleep(0.1)
+        return
+        
+    def getStartCommitNotifyFromTmqsim(self,cdbName='cdb'):
+        while 1:
+            tdSql.query("select * from %s.notifyinfo"%cdbName)
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            if tdSql.getRows() == 2 :
+                print(tdSql.getData(0, 1), tdSql.getData(1, 1))
+                if tdSql.getData(1, 1) == 1:
+                    break
+            time.sleep(0.1)
+        return
+
     def selectConsumeResult(self,expectRows,cdbName='cdb'):
         resultList=[]
         while 1:
@@ -72,7 +95,7 @@ class TDTestCase:
             if tdSql.getRows() == expectRows:
                 break
             else:
-                time.sleep(5)        
+                time.sleep(1)        
 
         for i in range(expectRows):
             tdLog.info ("ts: %s, consume id: %d, consume msgs: %d, consume rows: %d"%(tdSql.getData(i , 0), tdSql.getData(i , 1), tdSql.getData(i , 2), tdSql.getData(i , 3)))
@@ -207,7 +230,9 @@ class TDTestCase:
         showRow   = 1
         self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
 
-        time.sleep(2)
+        tdLog.info("wait the notify info of start consume")
+        self.getStartConsumeNotifyFromTmqsim()
+
         tdLog.info("pkill consume processor")
         if (platform.system().lower() == 'windows'):
             os.system("TASKKILL /F /IM tmq_sim.exe")
@@ -282,14 +307,17 @@ class TDTestCase:
         showRow   = 1
         self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
 
-        time.sleep(6)
+        # time.sleep(6)        
+        tdLog.info("start to wait commit notify")
+        self.getStartCommitNotifyFromTmqsim()
+
         tdLog.info("pkill consume processor")
         if (platform.system().lower() == 'windows'):
             os.system("TASKKILL /F /IM tmq_sim.exe")
         else:
             os.system('pkill tmq_sim')
-        expectRows = 0
+        # expectRows = 0
-        resultList = self.selectConsumeResult(expectRows)
+        # resultList = self.selectConsumeResult(expectRows)
 
         # wait for data ready
         prepareEnvThread.join()        

tests/test/c/tmqSim.c

@@ -34,6 +34,12 @@
 #define MAX_CONSUMER_THREAD_CNT (16)
 #define MAX_VGROUP_CNT          (32)
 
+typedef enum {
+    NOTIFY_CMD_START_CONSUM,
+	NOTIFY_CMD_START_COMMIT,
+	NOTIFY_CMD_ID_BUTT
+}NOTIFY_CMD_ID;
+
 typedef struct {
   TdThread thread;
   int32_t  consumerId;
@@ -67,6 +73,8 @@ typedef struct {
   int32_t rowsOfPerVgroups[MAX_VGROUP_CNT][2];  // [i][0]: vgroup id, [i][1]: rows of consume
   int64_t ts;
 
+  TAOS* taos;
+
 } SThreadInfo;
 
 typedef struct {
@@ -339,8 +347,37 @@ int queryDB(TAOS* taos, char* command) {
   return 0;
 }
 
+static void appNothing(void* param, TAOS_RES* res, int32_t numOfRows) {  
+}
+
+int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) {
+  char sqlStr[1024] = {0};
+
+  int64_t now = taosGetTimestampMs();
+
+  // schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int
+  sprintf(sqlStr, "insert into %s.notifyinfo values (%"PRId64", %d, %d)", 
+                   g_stConfInfo.cdbName, 
+                   now,
+                   cmdId,
+                   pInfo->consumerId);
+
+  taos_query_a(pInfo->taos, sqlStr, appNothing, NULL);
+
+  taosFprintfFile(g_fp, "notifyMainScript success, sql: %s\n", sqlStr);
+
+  return 0;
+}
+
+static int32_t g_once_commit_flag = 0;
 static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) {
-  pError("tmq_commit_cb_print() commit %d\n", code);
+  pError("tmq_commit_cb_print() commit %d\n", code);  
+
+  if (0 == g_once_commit_flag) {
+    g_once_commit_flag = 1;
+	notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT);
+  }
+  taosFprintfFile(g_fp, "tmq_commit_cb_print() be called\n");
 }
 
 void build_consumer(SThreadInfo* pInfo) {
@@ -353,7 +390,7 @@ void build_consumer(SThreadInfo* pInfo) {
 
   // tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName);
 
-  tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
+  tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, pInfo);
 
   // tmq_conf_set(conf, "group.id", "cgrp1");
   for (int32_t i = 0; i < pInfo->numOfKey; i++) {
@@ -392,9 +429,6 @@ void build_topic_list(SThreadInfo* pInfo) {
 int32_t saveConsumeResult(SThreadInfo* pInfo) {
   char sqlStr[1024] = {0};
 
-  TAOS* pConn = taos_connect(NULL, "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
-
   int64_t now = taosGetTimestampMs();
 
   // schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int
@@ -404,7 +438,7 @@ int32_t saveConsumeResult(SThreadInfo* pInfo) {
   char tmpString[128];
   taosFprintfFile(g_fp, "%s, consume id %d result: %s\n", getCurrentTimeString(tmpString), pInfo->consumerId, sqlStr);
 
-  TAOS_RES* pRes = taos_query(pConn, sqlStr);
+  TAOS_RES* pRes = taos_query(pInfo->taos, sqlStr);
   if (taos_errno(pRes) != 0) {
     pError("error in save consumeinfo, reason:%s\n", taos_errstr(pRes));
     taos_free_result(pRes);
@@ -413,38 +447,14 @@ int32_t saveConsumeResult(SThreadInfo* pInfo) {
 
   taos_free_result(pRes);
 
-#if 0
-  // vgroups
-  for (i = 0; i < pInfo->numOfVgroups; i++) {
-    // schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int
-    sprintf(sqlStr, "insert into %s.vgroup_%d values (%"PRId64", %d, %" PRId64 ", %" PRId64 ", %d)", 
-                     g_stConfInfo.cdbName,
-                     now,
-                     pInfo->consumerId, 
-                     pInfo->consumeMsgCnt, 
-                     pInfo->consumeRowCnt, 
-                     pInfo->checkresult);
-  
-    char tmpString[128];
-    taosFprintfFile(g_fp, "%s, consume id %d result: %s\n", getCurrentTimeString(tmpString), pInfo->consumerId ,sqlStr);
-  
-    TAOS_RES* pRes = taos_query(pConn, sqlStr);
-    if (taos_errno(pRes) != 0) {
-      pError("error in save consumeinfo, reason:%s\n", taos_errstr(pRes));
-      taos_free_result(pRes);
-      exit(-1);
-    }
-  
-    taos_free_result(pRes);
-  }
-#endif
-
   return 0;
 }
 
 void loop_consume(SThreadInfo* pInfo) {
   int32_t code;
 
+  int32_t once_flag = 0;
+
   int64_t totalMsgs = 0;
   int64_t totalRows = 0;
 
@@ -465,6 +475,11 @@ void loop_consume(SThreadInfo* pInfo) {
 
       totalMsgs++;
 
+	  if (0 == once_flag) {
+        once_flag = 1;
+		notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM);
+	  }
+
       if (totalRows >= pInfo->expectMsgCnt) {
         char tmpString[128];
         taosFprintfFile(g_fp, "%s over than expect rows, so break consume\n", getCurrentTimeString(tmpString));
@@ -489,6 +504,12 @@ void* consumeThreadFunc(void* param) {
 
   SThreadInfo* pInfo = (SThreadInfo*)param;
 
+  pInfo->taos = taos_connect(NULL, "root", "taosdata", NULL, 0);
+  if (pInfo->taos == NULL) {
+    taosFprintfFile(g_fp, "taos_connect() fail, can not notify and save consume result to main scripte\n");
+	exit(-1);
+  }
+
   build_consumer(pInfo);
   build_topic_list(pInfo);
   if ((NULL == pInfo->tmq) || (NULL == pInfo->topicList)) {
@@ -508,7 +529,6 @@ void* consumeThreadFunc(void* param) {
   loop_consume(pInfo);
 
   if (pInfo->ifManualCommit) {
-    taosFprintfFile(g_fp, "tmq_commit() manual commit when consume end.\n");
     pPrint("tmq_commit() manual commit when consume end.\n");
     /*tmq_commit(pInfo->tmq, NULL, 0);*/
     tmq_commit_sync(pInfo->tmq, NULL);
@@ -539,6 +559,9 @@ void* consumeThreadFunc(void* param) {
     taosFprintfFile(g_fp, "vgroups: %04d, rows: %d\n", pInfo->rowsOfPerVgroups[i][0], pInfo->rowsOfPerVgroups[i][1]);
   }
 
+  taos_close(pInfo->taos);
+  pInfo->taos = NULL;
+
   return NULL;
 }
 

tools/taosadapter

@@ -1 +1 @@
-Subproject commit 29926478edd87533a043f91c1a9ed0e27671e626
+Subproject commit c1dc11f3cae64adf31dbd9a954ef8372d1e8f671