Merge branch '3.0' of https://github.com/taosdata/TDengine into enh/tsdb_optimize

docs/en/12-taos-sql/04-stable.md

@@ -51,6 +51,11 @@ DESCRIBE [db_name.]stb_name;
 
 ### View tag information for all child tables in the supertable
 
+```
+SHOW TABLE TAGS FROM table_name [FROM db_name];
+SHOW TABLE TAGS FROM [db_name.]table_name;
+```
+
 ```
 taos> SHOW TABLE TAGS FROM st1;
              tbname             |     id      |         loc          |

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

@@ -101,6 +101,7 @@ Note: TDengine Enterprise Edition only.
 
 ```sql
 SHOW INDEXES FROM tbl_name [FROM db_name];
+SHOW INDEXES FROM [db_name.]tbl_name;
 ```
 
 Shows indices that have been created.
@@ -326,6 +327,7 @@ Note that only the information about the data blocks in the data file will be di
 
 ```sql
 SHOW TAGS FROM child_table_name [FROM db_name];
+SHOW TAGS FROM [db_name.]child_table_name;
 ```
 
 Shows all tag information in a subtable.

docs/en/12-taos-sql/27-index.md

@@ -43,6 +43,7 @@ DROP INDEX index_name;
 ````sql
 ```sql
 SHOW INDEXES FROM tbl_name [FROM db_name];
+SHOW INDEXES FROM [db_name.]tbl_name ;
 ````
 
 Shows indices that have been created for the specified database or table.

docs/en/14-reference/03-connector/04-java.mdx

@@ -36,8 +36,8 @@ REST connection supports all platforms that can run Java.
 
 | taos-jdbcdriver version |                                                                   major changes                                                                    | TDengine version |
 | :---------------------: | :------------------------------------------------------------------------------------------------------------------------------------------------: | :--------------: |
-|          3.2.4          |                     Subscription add the enable.auto.commit parameter and the unsubscribe() method in the WebSocket connection                     | 3.0.5.0 or later |
-|          3.2.3          |                                                 Fixed resultSet data parsing failure in some cases                                                 | 3.0.5.0 or later |
+|          3.2.4          |                     Subscription add the enable.auto.commit parameter and the unsubscribe() method in the WebSocket connection                     |        -         |
+|          3.2.3          |                                                 Fixed resultSet data parsing failure in some cases                                                 |        -         |
 |          3.2.2          |                                                           Subscription add seek function                                                           | 3.0.5.0 or later |
 |          3.2.1          |                                      JDBC REST connection supports schemaless/prepareStatement over WebSocket                                      | 3.0.3.0 or later |
 |          3.2.0          |                                                          This version has been deprecated                                                          |        -         |
@@ -1019,11 +1019,13 @@ while(true) {
 #### Assignment subscription Offset
 
 ```java
+// get offset
 long position(TopicPartition partition) throws SQLException;
 Map<TopicPartition, Long> position(String topic) throws SQLException;
 Map<TopicPartition, Long> beginningOffsets(String topic) throws SQLException;
 Map<TopicPartition, Long> endOffsets(String topic) throws SQLException;
 
+// Overrides the fetch offsets that the consumer will use on the next poll(timeout).
 void seek(TopicPartition partition, long offset) throws SQLException;
 ```
 

docs/en/14-reference/03-connector/07-python.mdx

@@ -87,9 +87,9 @@ TDengine currently supports timestamp, number, character, Boolean type, and the
 |NCHAR|str|
 |JSON|str|
 
-## Installation
+## Installation Steps
 
-### Preparation
+### Pre-installation preparation
 
 1. Install Python. The recent taospy package requires Python 3.6.2+. The earlier versions of taospy require Python 3.7+.  The taos-ws-py package requires Python 3.7+. If Python is not available on your system, refer to the [Python BeginnersGuide](https://wiki.python.org/moin/BeginnersGuide/Download) to install it.
 2. Install [pip](https://pypi.org/project/pip/). In most cases, the Python installer comes with the pip utility. If not, please refer to [pip documentation](https://pip.pypa.io/en/stable/installation/) to install it.
@@ -275,7 +275,7 @@ Transfer-Encoding: chunked
 </TabItem>
 </Tabs>
 
-### Using connectors to establish connections
+### Specify the Host and Properties to get the connection
 
 The following example code assumes that TDengine is installed locally and that the default configuration is used for both FQDN and serverPort.
 
@@ -331,7 +331,69 @@ The parameter of `connect()` is the url of TDengine, and the protocol is `taosws
 </TabItem>
 </Tabs>
 
-## Example program
+### Priority of configuration parameters
+
+If the configuration parameters are duplicated in the parameters or client configuration file, the priority of the parameters, from highest to lowest, are as follows:
+
+1. Parameters in `connect` function.
+2. the configuration file taos.cfg of the TDengine client driver when using a native connection.
+
+## Usage examples
+
+### Create database and tables
+
+<Tabs defaultValue="rest">
+<TabItem value="native" label="native connection">
+
+```python
+conn = taos.connect()
+# Execute a sql, ignore the result set, just get affected rows. It's useful for DDL and DML statement.
+conn.execute("DROP DATABASE IF EXISTS test")
+conn.execute("CREATE DATABASE test")
+# change database. same as execute "USE db"
+conn.select_db("test")
+conn.execute("CREATE STABLE weather(ts TIMESTAMP, temperature FLOAT) TAGS (location INT)")
+```
+
+</TabItem>
+
+<TabItem value="rest" label="REST connection">
+
+```python
+conn = taosrest.connect(url="http://localhost:6041")
+# Execute a sql, ignore the result set, just get affected rows. It's useful for DDL and DML statement.
+conn.execute("DROP DATABASE IF EXISTS test")
+conn.execute("CREATE DATABASE test")
+conn.execute("USE test")
+conn.execute("CREATE STABLE weather(ts TIMESTAMP, temperature FLOAT) TAGS (location INT)")
+```
+
+</TabItem>
+
+<TabItem value="websocket" label="WebSocket connection">
+
+```python
+conn = taosws.connect(url="ws://localhost:6041")
+# Execute a sql, ignore the result set, just get affected rows. It's useful for DDL and DML statement.
+conn.execute("DROP DATABASE IF EXISTS test")
+conn.execute("CREATE DATABASE test")
+conn.execute("USE test")
+conn.execute("CREATE STABLE weather(ts TIMESTAMP, temperature FLOAT) TAGS (location INT)")
+```
+
+</TabItem>
+</Tabs>
+
+### Insert data
+
+```python
+conn.execute("INSERT INTO t1 USING weather TAGS(1) VALUES (now, 23.5) (now+1m, 23.5) (now+2m, 24.4)")
+```
+
+:::
+now is an internal function. The default is the current time of the client's computer. now + 1s represents the current time of the client plus 1 second, followed by the number representing the unit of time: a (milliseconds), s (seconds), m (minutes), h (hours), d (days), w (weeks), n (months), y (years).
+:::
+
 
 ### Basic Usage
 
@@ -453,7 +515,7 @@ The `query` method of the `TaosConnection` class can be used to query data and r
 </TabItem>
 </Tabs>
 
-### Usage with req_id
+### Execute SQL with reqId
 
 By using the optional req_id parameter, you can specify a request ID that can be used for tracing.
 
@@ -553,171 +615,138 @@ As the way to connect introduced above but add `req_id` argument.
 </TabItem>
 </Tabs>
 
-### Subscription
+### Writing data via parameter binding
 
-Connector support data subscription. For more information about subscroption, please refer to [Data Subscription](../../../develop/tmq/).
+The Python connector provides a parameter binding api for inserting data. Similar to most databases, TDengine currently only supports the question mark `?` to indicate the parameters to be bound.
 
-<Tabs defaultValue="native">
+<Tabs>
 <TabItem value="native" label="native connection">
 
-The `consumer` in the connector contains the subscription api. 
-
-##### Create Consumer
-
-The syntax for creating a consumer is `consumer = Consumer(configs)`. For more subscription api parameters, please refer to [Data Subscription](../../../develop/tmq/).
+##### Create Stmt
 
-```python
-from taos.tmq import Consumer
+Call the `statement` method in `Connection` to create the `stmt` for parameter binding.
 
-consumer = Consumer({"group.id": "local", "td.connect.ip": "127.0.0.1"})
 ```
+import taos
 
-##### Subscribe topics
-
-The `subscribe` function is used to subscribe to a list of topics.
-
-```python
-consumer.subscribe(['topic1', 'topic2'])
+conn = taos.connect()
+stmt = conn.statement("insert into log values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
 ```
 
-##### Consume
-
-The `poll` function is used to consume data in tmq. The parameter of the `poll` function is a value of type float representing the timeout in seconds. It returns a `Message` before timing out, or `None` on timing out. You have to handle error messages in response data.
+##### parameter binding
 
-```python
-while True:
-    res = consumer.poll(1)
-    if not res:
-        continue
-    err = res.error()
-    if err is not None:
-        raise err
-    val = res.value()
+Call the `new_multi_binds` function to create the parameter list for parameter bindings.
 
-    for block in val:
-        print(block.fetchall())
 ```
-
-##### assignment
-
-The `assignment` function is used to get the assignment of the topic. 
-
-```python
-assignments = consumer.assignment()
+params = new_multi_binds(16)
+params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
+params[1].bool((True, None, False))
+params[2].tinyint([-128, -128, None])  # -128 is tinyint null
+params[3].tinyint([0, 127, None])
+params[4].smallint([3, None, 2])
+params[5].int([3, 4, None])
+params[6].bigint([3, 4, None])
+params[7].tinyint_unsigned([3, 4, None])
+params[8].smallint_unsigned([3, 4, None])
+params[9].int_unsigned([3, 4, None])
+params[10].bigint_unsigned([3, 4, None])
+params[11].float([3, None, 1])
+params[12].double([3, None, 1.2])
+params[13].binary(["abc", "dddafadfadfadfadfa", None])
+params[14].nchar(["涛思数据", None, "a long string with 中文字符"])
+params[15].timestamp([None, None, 1626861392591])
 ```
 
-##### Seek
-
-The `seek` function is used to reset the assignment of the topic.
+Call the `bind_param` (for a single row) method or the `bind_param_batch` (for multiple rows) method to set the values.
 
-```python
-tp = TopicPartition(topic='topic1', partition=0, offset=0)
-consumer.seek(tp)
+```
+stmt.bind_param_batch(params)
 ```
 
-##### After consuming data
+##### execute sql
 
-You should unsubscribe to the topics and close the consumer after consuming.
+Call `execute` method to execute sql.
 
-```python
-consumer.unsubscribe()
-consumer.close()
+```
+stmt.execute()
 ```
 
-##### Tmq subscription example
+##### Close Stmt
 
-```python
-{{#include docs/examples/python/tmq_example.py}}
+```
+stmt.close()
 ```
 
-##### assignment and seek example
+##### Example
 
 ```python
-{{#include docs/examples/python/tmq_assignment_example.py:taos_get_assignment_and_seek_demo}}
+{{#include docs/examples/python/stmt_example.py}}
 ```
-
 </TabItem>
 
 <TabItem value="websocket" label="WebSocket connection">
 
-In addition to native connections, the connector also supports subscriptions via websockets.
-
-##### Create Consumer
+##### Create Stmt
 
-The syntax for creating a consumer is "consumer = consumer = Consumer(conf=configs)". You need to specify that the `td.connect.websocket.scheme` parameter is set to "ws" in the configuration. For more subscription api parameters, please refer to [Data Subscription](../../../develop/tmq/#create-a-consumer).
+Call the `statement` method in `Connection` to create the `stmt` for parameter binding.
 
-```python
+```
 import taosws
 
-consumer = taosws.(conf={"group.id": "local", "td.connect.websocket.scheme": "ws"})
+conn = taosws.connect('taosws://localhost:6041/test')
+stmt = conn.statement()
 ```
 
-##### subscribe topics
+##### Prepare sql
 
-The `subscribe` function is used to subscribe to a list of topics.
+Call `prepare` method in stmt to prepare sql.
 
-```python
-consumer.subscribe(['topic1', 'topic2'])
 ```
-
-##### Consume
-
-The `poll` function is used to consume data in tmq. The parameter of the `poll` function is a value of type float representing the timeout in seconds. It returns a `Message` before timing out, or `None` on timing out. You have to handle error messages in response data.
-
-```python
-while True:
-    res = consumer.poll(timeout=1.0)
-    if not res:
-        continue
-    err = res.error()
-    if err is not None:
-        raise err
-    for block in message:
-        for row in block:
-            print(row)
+stmt.prepare("insert into t1 values (?, ?, ?, ?)")
 ```
 
-##### assignment
+##### parameter binding
 
-The `assignment` function is used to get the assignment of the topic. 
+Call the `bind_param` method to bind parameters.
 
-```python
-assignments = consumer.assignment()
+```
+stmt.bind_param([
+    taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+    taosws.ints_to_column([1, 2, 3, 4]),
+    taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+    taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+])
 ```
 
-##### Seek
-
-The `seek` function is used to reset the assignment of the topic.
+Call the `add_batch` method to add parameters to the batch.
 
-```python
-consumer.seek(topic='topic1', partition=0, offset=0)
+```
+stmt.add_batch()
 ```
 
-##### After consuming data
+##### execute sql
 
-You should unsubscribe to the topics and close the consumer after consuming.
+Call `execute` method to execute sql.
 
-```python
-consumer.unsubscribe()
-consumer.close()
+```
+stmt.execute()
 ```
 
-##### Subscription example
+##### Close Stmt
 
-```python
-{{#include docs/examples/python/tmq_websocket_example.py}}
+```
+stmt.close()
 ```
 
-##### Assignment and seek example
+##### Example
 
 ```python
-{{#include docs/examples/python/tmq_websocket_assgnment_example.py:taosws_get_assignment_and_seek_demo}}
+{{#include docs/examples/python/stmt_websocket_example.py}}
 ```
-
 </TabItem>
 </Tabs>
 
-### Schemaless Insert
+### Schemaless Writing
 
 Connector support schemaless insert.
 
@@ -767,134 +796,211 @@ Connector support schemaless insert.
 </TabItem>
 </Tabs>
 
-### Parameter Binding
+### Schemaless with reqId
 
-The Python connector provides a parameter binding api for inserting data. Similar to most databases, TDengine currently only supports the question mark `?` to indicate the parameters to be bound.
+There is a optional parameter called `req_id` in `schemaless_insert` and `schemaless_insert_raw` method. This reqId can be used to request link tracing.
+
+```python
+{{#include docs/examples/python/schemaless_insert_req_id.py}}
+```
+
+```python
+{{#include docs/examples/python/schemaless_insert_raw_req_id.py}}
+```
+
+### Data Subscription
+
+Connector support data subscription. For more information about subscroption, please refer to [Data Subscription](../../../develop/tmq/).
+
+#### Create a Topic
+
+To create topic, please refer to [Data Subscription](../../../develop/tmq/#create-a-topic).
+
+#### Create a Consumer
+
+<Tabs defaultValue="native">
 
-<Tabs>
 <TabItem value="native" label="native connection">
 
-##### Create Stmt
+The consumer in the connector contains the subscription api. The syntax for creating a consumer is consumer = Consumer(configs). For more subscription api parameters, please refer to [Data Subscription](../../../develop/tmq/#create-a-consumer).
 
-Call the `statement` method in `Connection` to create the `stmt` for parameter binding.
+```python
+from taos.tmq import Consumer
 
+consumer = Consumer({"group.id": "local", "td.connect.ip": "127.0.0.1"})
 ```
-import taos
+</TabItem>
 
-conn = taos.connect()
-stmt = conn.statement("insert into log values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
-```
+<TabItem value="websocket" label="WebSocket connection">
 
-##### parameter binding
+In addition to native connections, the connector also supports subscriptions via websockets.
 
-Call the `new_multi_binds` function to create the parameter list for parameter bindings.
+The syntax for creating a consumer is "consumer = consumer = Consumer(conf=configs)". You need to specify that the `td.connect.websocket.scheme` parameter is set to "ws" in the configuration. For more subscription api parameters, please refer to [Data Subscription](../../../develop/tmq/#create-a-consumer).
 
-```
-params = new_multi_binds(16)
-params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
-params[1].bool((True, None, False))
-params[2].tinyint([-128, -128, None])  # -128 is tinyint null
-params[3].tinyint([0, 127, None])
-params[4].smallint([3, None, 2])
-params[5].int([3, 4, None])
-params[6].bigint([3, 4, None])
-params[7].tinyint_unsigned([3, 4, None])
-params[8].smallint_unsigned([3, 4, None])
-params[9].int_unsigned([3, 4, None])
-params[10].bigint_unsigned([3, 4, None])
-params[11].float([3, None, 1])
-params[12].double([3, None, 1.2])
-params[13].binary(["abc", "dddafadfadfadfadfa", None])
-params[14].nchar(["涛思数据", None, "a long string with 中文字符"])
-params[15].timestamp([None, None, 1626861392591])
+```python
+import taosws
+
+consumer = taosws.(conf={"group.id": "local", "td.connect.websocket.scheme": "ws"})
 ```
 
-Call the `bind_param` (for a single row) method or the `bind_param_batch` (for multiple rows) method to set the values.
+</TabItem>
+</Tabs>
 
-```
-stmt.bind_param_batch(params)
-```
+#### Subscribe to a Topic
 
-##### execute sql
+<Tabs defaultValue="native">
 
-Call `execute` method to execute sql.
+<TabItem value="native" label="native connection">
 
-```
-stmt.execute()
+The `subscribe` function is used to subscribe to a list of topics.
+
+```python
+consumer.subscribe(['topic1', 'topic2'])
 ```
 
-##### Close Stmt
+</TabItem>
+<TabItem value="websocket" label="WebSocket connection">
+
+The `subscribe` function is used to subscribe to a list of topics.
 
+```python
+consumer.subscribe(['topic1', 'topic2'])
 ```
-stmt.close()
+
+</TabItem>
+</Tabs>
+
+#### Consume messages
+
+<Tabs defaultValue="native">
+
+<TabItem value="native" label="native connection">
+
+The `poll` function is used to consume data in tmq. The parameter of the `poll` function is a value of type float representing the timeout in seconds. It returns a `Message` before timing out, or `None` on timing out. You have to handle error messages in response data.
+
+```python
+while True:
+    res = consumer.poll(1)
+    if not res:
+        continue
+    err = res.error()
+    if err is not None:
+        raise err
+    val = res.value()
+
+    for block in val:
+        print(block.fetchall())
 ```
 
-##### Example
+</TabItem>
+<TabItem value="websocket" label="WebSocket connection">
+
+The `poll` function is used to consume data in tmq. The parameter of the `poll` function is a value of type float representing the timeout in seconds. It returns a `Message` before timing out, or `None` on timing out. You have to handle error messages in response data.
 
 ```python
-{{#include docs/examples/python/stmt_example.py}}
+while True:
+    res = consumer.poll(timeout=1.0)
+    if not res:
+        continue
+    err = res.error()
+    if err is not None:
+        raise err
+    for block in message:
+        for row in block:
+            print(row)
 ```
+
 </TabItem>
+</Tabs>
 
-<TabItem value="websocket" label="WebSocket connection">
+#### Assignment subscription Offset
 
-##### Create Stmt
+<Tabs defaultValue="native">
 
-Call the `statement` method in `Connection` to create the `stmt` for parameter binding.
+<TabItem value="native" label="native connection">
 
+The `assignment` function is used to get the assignment of the topic. 
+
+```python
+assignments = consumer.assignment()
 ```
-import taosws
 
-conn = taosws.connect('taosws://localhost:6041/test')
-stmt = conn.statement()
+The `seek` function is used to reset the assignment of the topic.
+
+```python
+tp = TopicPartition(topic='topic1', partition=0, offset=0)
+consumer.seek(tp)
 ```
 
-##### Prepare sql
+</TabItem>
+<TabItem value="websocket" label="WebSocket connection">
 
-Call `prepare` method in stmt to prepare sql.
+The `assignment` function is used to get the assignment of the topic. 
 
+```python
+assignments = consumer.assignment()
 ```
-stmt.prepare("insert into t1 values (?, ?, ?, ?)")
+
+The `seek` function is used to reset the assignment of the topic.
+
+```python
+consumer.seek(topic='topic1', partition=0, offset=0)
 ```
 
-##### parameter binding
+</TabItem>
+</Tabs>
 
-Call the `bind_param` method to bind parameters.
+#### Close subscriptions
 
-```
-stmt.bind_param([
-    taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
-    taosws.ints_to_column([1, 2, 3, 4]),
-    taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
-    taosws.varchar_to_column(['a', 'b', 'c', 'd']),
-])
-```
+<Tabs defaultValue="native">
 
-Call the `add_batch` method to add parameters to the batch.
+<TabItem value="native" label="native connection">
 
-```
-stmt.add_batch()
+You should unsubscribe to the topics and close the consumer after consuming.
+
+```python
+consumer.unsubscribe()
+consumer.close()
 ```
 
-##### execute sql
+</TabItem>
+<TabItem value="websocket" label="WebSocket connection">
 
-Call `execute` method to execute sql.
+You should unsubscribe to the topics and close the consumer after consuming.
 
-```
-stmt.execute()
+```python
+consumer.unsubscribe()
+consumer.close()
 ```
 
-##### Close Stmt
+</TabItem>
+</Tabs>
+
+#### Full Sample Code
+
+<Tabs defaultValue="native">
 
+<TabItem value="native" label="native connection">
+
+```python
+{{#include docs/examples/python/tmq_example.py}}
 ```
-stmt.close()
+
+```python
+{{#include docs/examples/python/tmq_assignment_example.py:taos_get_assignment_and_seek_demo}}
 ```
 
-##### Example
+</TabItem>
+<TabItem value="websocket" label="WebSocket connection">
 
 ```python
-{{#include docs/examples/python/stmt_websocket_example.py}}
+{{#include docs/examples/python/tmq_websocket_example.py}}
+```
+
+```python
+{{#include docs/examples/python/tmq_websocket_assgnment_example.py:taosws_get_assignment_and_seek_demo}}
 ```
+
 </TabItem>
 </Tabs>
 

docs/zh/08-connector/14-java.mdx

@@ -1022,11 +1022,13 @@ while(true) {
 #### 指定订阅 Offset
 
 ```java
+// 获取 offset
 long position(TopicPartition partition) throws SQLException;
 Map<TopicPartition, Long> position(String topic) throws SQLException;
 Map<TopicPartition, Long> beginningOffsets(String topic) throws SQLException;
 Map<TopicPartition, Long> endOffsets(String topic) throws SQLException;
 
+// 指定下一次 poll 中使用的 offset
 void seek(TopicPartition partition, long offset) throws SQLException;
 ```
 

docs/zh/08-connector/30-python.mdx

@@ -70,7 +70,7 @@ Python Connector 的所有数据库操作如果出现异常，都会直接抛出
 {{#include docs/examples/python/handle_exception.py}}
 ```
 
-TDengine DataType 和 Python DataType
+## TDengine DataType 和 Python DataType
 
 TDengine 目前支持时间戳、数字、字符、布尔类型，与 Python 对应类型转换如下：
 
@@ -276,7 +276,7 @@ Transfer-Encoding: chunked
 </TabItem>
 </Tabs>
 
-### 使用连接器建立连接
+### 指定 Host 和 Properties 获取连接
 
 以下示例代码假设 TDengine 安装在本机， 且 FQDN 和 serverPort 都使用了默认配置。
 
@@ -332,8 +332,69 @@ Transfer-Encoding: chunked
 </TabItem>
 </Tabs>
 
+### 配置参数的优先级
+
+如果配置参数在参数和客户端配置文件中有重复，则参数的优先级由高到低分别如下：
+
+1. 连接参数
+2. 使用原生连接时，TDengine 客户端驱动的配置文件 taos.cfg
+
 ## 使用示例
 
+### 创建数据库和表
+
+<Tabs defaultValue="rest">
+<TabItem value="native" label="原生连接">
+
+```python
+conn = taos.connect()
+# Execute a sql, ignore the result set, just get affected rows. It's useful for DDL and DML statement.
+conn.execute("DROP DATABASE IF EXISTS test")
+conn.execute("CREATE DATABASE test")
+# change database. same as execute "USE db"
+conn.select_db("test")
+conn.execute("CREATE STABLE weather(ts TIMESTAMP, temperature FLOAT) TAGS (location INT)")
+```
+
+</TabItem>
+
+<TabItem value="rest" label="REST 连接">
+
+```python
+conn = taosrest.connect(url="http://localhost:6041")
+# Execute a sql, ignore the result set, just get affected rows. It's useful for DDL and DML statement.
+conn.execute("DROP DATABASE IF EXISTS test")
+conn.execute("CREATE DATABASE test")
+conn.execute("USE test")
+conn.execute("CREATE STABLE weather(ts TIMESTAMP, temperature FLOAT) TAGS (location INT)")
+```
+
+</TabItem>
+
+<TabItem value="websocket" label="WebSocket 连接">
+
+```python
+conn = taosws.connect(url="ws://localhost:6041")
+# Execute a sql, ignore the result set, just get affected rows. It's useful for DDL and DML statement.
+conn.execute("DROP DATABASE IF EXISTS test")
+conn.execute("CREATE DATABASE test")
+conn.execute("USE test")
+conn.execute("CREATE STABLE weather(ts TIMESTAMP, temperature FLOAT) TAGS (location INT)")
+```
+
+</TabItem>
+</Tabs>
+
+### 插入数据
+
+```python
+conn.execute("INSERT INTO t1 USING weather TAGS(1) VALUES (now, 23.5) (now+1m, 23.5) (now+2m, 24.4)")
+```
+
+:::
+now 为系统内部函数，默认为客户端所在计算机当前时间。 now + 1s 代表客户端当前时间往后加 1 秒，数字后面代表时间单位：a(毫秒)，s(秒)，m(分)，h(小时)，d(天)，w(周)，n(月)，y(年)。
+:::
+
 ### 基本使用
 
 <Tabs defaultValue="rest">
@@ -372,7 +433,6 @@ Transfer-Encoding: chunked
 
 :::note
 TaosCursor 类使用原生连接进行写入、查询操作。在客户端多线程的场景下，这个游标实例必须保持线程独享，不能跨线程共享使用，否则会导致返回结果出现错误。
-
 :::
 
 </TabItem>
@@ -455,7 +515,7 @@ RestClient 类是对于 REST API 的直接封装。它只包含一个 sql() 方
 </TabItem>
 </Tabs>
 
-### 与 req_id 一起使用
+### 执行带有 reqId 的 SQL
 
 使用可选的 req_id 参数，指定请求 id，可以用于 tracing
 
@@ -556,171 +616,138 @@ RestClient 类是对于 REST API 的直接封装。它只包含一个 sql() 方
 </TabItem>
 </Tabs>
 
-### 数据订阅
+### 通过参数绑定写入数据
 
-连接器支持数据订阅功能，数据订阅功能请参考 [数据订阅文档](../../develop/tmq/)。
+TDengine 的 Python 连接器支持参数绑定风格的 Prepare API 方式写入数据，和大多数数据库类似，目前仅支持用 `?` 来代表待绑定的参数。
 
-<Tabs defaultValue="native">
+<Tabs>
 <TabItem value="native" label="原生连接">
 
-`Consumer` 提供了 Python 连接器订阅 TMQ 数据的 API。
-
-##### 创建 Consumer
-
-创建 Consumer 语法为 `consumer = Consumer(configs)`，参数定义请参考 [数据订阅文档](../../develop/tmq/#%E5%88%9B%E5%BB%BA%E6%B6%88%E8%B4%B9%E8%80%85-consumer)。
+##### 创建 stmt
 
-```python
-from taos.tmq import Consumer
+Python 连接器的 `Connection` 提供了 `statement` 方法用于创建参数绑定对象 stmt，该方法接收 sql 字符串作为参数，sql 字符串目前仅支持用 `?` 来代表绑定的参数。
 
-consumer = Consumer({"group.id": "local", "td.connect.ip": "127.0.0.1"})
 ```
+import taos
 
-##### 订阅 topics
-
-Consumer API 的 `subscribe` 方法用于订阅 topics，consumer 支持同时订阅多个 topic。
-
-```python
-consumer.subscribe(['topic1', 'topic2'])
+conn = taos.connect()
+stmt = conn.statement("insert into log values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
 ```
 
-##### 消费数据
-
-Consumer API 的 `poll` 方法用于消费数据，`poll` 方法接收一个 float 类型的超时时间，超时时间单位为秒（s），`poll` 方法在超时之前返回一条 Message 类型的数据或超时返回 `None`。消费者必须通过 Message 的 `error()` 方法校验返回数据的 error 信息。
+##### 参数绑定
 
-```python
-while True:
-    res = consumer.poll(1)
-    if not res:
-        continue
-    err = res.error()
-    if err is not None:
-        raise err
-    val = res.value()
+调用 `new_multi_binds` 函数创建 params 列表，用于参数绑定。
 
-    for block in val:
-        print(block.fetchall())
+```
+params = new_multi_binds(16)
+params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
+params[1].bool((True, None, False))
+params[2].tinyint([-128, -128, None])  # -128 is tinyint null
+params[3].tinyint([0, 127, None])
+params[4].smallint([3, None, 2])
+params[5].int([3, 4, None])
+params[6].bigint([3, 4, None])
+params[7].tinyint_unsigned([3, 4, None])
+params[8].smallint_unsigned([3, 4, None])
+params[9].int_unsigned([3, 4, None])
+params[10].bigint_unsigned([3, 4, None])
+params[11].float([3, None, 1])
+params[12].double([3, None, 1.2])
+params[13].binary(["abc", "dddafadfadfadfadfa", None])
+params[14].nchar(["涛思数据", None, "a long string with 中文字符"])
+params[15].timestamp([None, None, 1626861392591])
 ```
 
-##### 获取消费进度
-
-Consumer API 的 `assignment` 方法用于获取 Consumer 订阅的所有 topic 的消费进度，返回结果类型为 TopicPartition 列表。
+调用 stmt 的 `bind_param` 以单行的方式设置 values 或 `bind_param_batch` 以多行的方式设置 values 方法绑定参数。
 
-```python
-assignments = consumer.assignment()
+```
+stmt.bind_param_batch(params)
 ```
 
-##### 指定订阅 Offset
+##### 执行 sql
 
-Consumer API 的 `seek` 方法用于重置 Consumer 的消费进度到指定位置，方法参数类型为 TopicPartition。
+调用 stmt 的 `execute` 方法执行 sql
 
-```python
-tp = TopicPartition(topic='topic1', partition=0, offset=0)
-consumer.seek(tp)
+```
+stmt.execute()
 ```
 
-##### 关闭订阅
+##### 关闭 stmt
 
-消费结束后，应当取消订阅，并关闭 Consumer。
+最后需要关闭 stmt。
 
-```python
-consumer.unsubscribe()
-consumer.close()
 ```
-
-##### 完整示例
-
-```python
-{{#include docs/examples/python/tmq_example.py}}
+stmt.close()
 ```
 
-##### 获取和重置消费进度示例代码
+##### 示例代码
 
 ```python
-{{#include docs/examples/python/tmq_assignment_example.py:taos_get_assignment_and_seek_demo}}
+{{#include docs/examples/python/stmt_example.py}}
 ```
-
 </TabItem>
 
 <TabItem value="websocket" label="WebSocket 连接">
 
-除了原生的连接方式，Python 连接器还支持通过 websocket 订阅 TMQ 数据，使用 websocket 方式订阅 TMQ 数据需要安装 `taos-ws-py`。
-
-taosws `Consumer` API 提供了基于 Websocket 订阅 TMQ 数据的 API。
-
-##### 创建 Consumer
+##### 创建 stmt
 
-创建 Consumer 语法为 `consumer = Consumer(conf=configs)`，使用时需要指定 `td.connect.websocket.scheme` 参数值为 "ws"，参数定义请参考 [数据订阅文档](../../develop/tmq/#%E5%88%9B%E5%BB%BA%E6%B6%88%E8%B4%B9%E8%80%85-consumer)。
+Python WebSocket 连接器的 `Connection` 提供了 `statement` 方法用于创建参数绑定对象 stmt，该方法接收 sql 字符串作为参数，sql 字符串目前仅支持用 `?` 来代表绑定的参数。
 
-```python
+```
 import taosws
 
-consumer = taosws.(conf={"group.id": "local", "td.connect.websocket.scheme": "ws"})
+conn = taosws.connect('taosws://localhost:6041/test')
+stmt = conn.statement()
 ```
 
-##### 订阅 topics
+##### 解析 sql
 
-Consumer API 的 `subscribe` 方法用于订阅 topics，consumer 支持同时订阅多个 topic。
+调用 stmt 的 `prepare` 方法来解析 insert 语句。
 
-```python
-consumer.subscribe(['topic1', 'topic2'])
+```
+stmt.prepare("insert into t1 values (?, ?, ?, ?)")
 ```
 
-##### 消费数据
+##### 参数绑定
 
-Consumer API 的 `poll` 方法用于消费数据，`poll` 方法接收一个 float 类型的超时时间，超时时间单位为秒（s），`poll` 方法在超时之前返回一条 Message 类型的数据或超时返回 `None`。消费者必须通过 Message 的 `error()` 方法校验返回数据的 error 信息。
+调用 stmt 的 `bind_param` 方法绑定参数。
 
-```python
-while True:
-    res = consumer.poll(timeout=1.0)
-    if not res:
-        continue
-    err = res.error()
-    if err is not None:
-        raise err
-    for block in message:
-        for row in block:
-            print(row)
+```
+stmt.bind_param([
+    taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
+    taosws.ints_to_column([1, 2, 3, 4]),
+    taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
+    taosws.varchar_to_column(['a', 'b', 'c', 'd']),
+])
 ```
 
-##### 获取消费进度
-
-Consumer API 的 `assignment` 方法用于获取 Consumer 订阅的所有 topic 的消费进度，返回结果类型为 TopicPartition 列表。
+调用 stmt 的 `add_batch` 方法，将参数加入批处理。
 
-```python
-assignments = consumer.assignment()
+```
+stmt.add_batch()
 ```
 
-##### 重置消费进度
+##### 执行 sql
 
-Consumer API 的 `seek` 方法用于重置 Consumer 的消费进度到指定位置。
+调用 stmt 的 `execute` 方法执行 sql
 
-```python
-consumer.seek(topic='topic1', partition=0, offset=0)
+```
+stmt.execute()
 ```
 
-##### 结束消费
+##### 关闭 stmt
 
-消费结束后，应当取消订阅，并关闭 Consumer。
+最后需要关闭 stmt。
 
-```python
-consumer.unsubscribe()
-consumer.close()
 ```
-
-##### tmq 订阅示例代码
-
-```python
-{{#include docs/examples/python/tmq_websocket_example.py}}
+stmt.close()
 ```
 
-连接器提供了 `assignment` 接口，用于获取 topic assignment 的功能，可以查询订阅的 topic 的消费进度，并提供 `seek` 接口，用于重置 topic 的消费进度。
-
-##### 获取和重置消费进度示例代码
+##### 示例代码
 
 ```python
-{{#include docs/examples/python/tmq_websocket_assgnment_example.py:taosws_get_assignment_and_seek_demo}}
+{{#include docs/examples/python/stmt_websocket_example.py}}
 ```
-
 </TabItem>
 </Tabs>
 
@@ -774,138 +801,211 @@ consumer.close()
 </TabItem>
 </Tabs>
 
-### 通过参数绑定写入数据
+### 执行带有 reqId 的无模式写入
 
-TDengine 的 Python 连接器支持参数绑定风格的 Prepare API 方式写入数据，和大多数数据库类似，目前仅支持用 `?` 来代表待绑定的参数。
+连接器的 `schemaless_insert` 和 `schemaless_insert_raw` 方法支持 `req_id` 可选参数，此 `req_Id` 可用于请求链路追踪。
+
+```python
+{{#include docs/examples/python/schemaless_insert_req_id.py}}
+```
+
+```python
+{{#include docs/examples/python/schemaless_insert_raw_req_id.py}}
+```
+
+### 数据订阅
+
+连接器支持数据订阅功能，数据订阅功能请参考 [数据订阅文档](../../develop/tmq/)。
+
+#### 创建 Topic
+
+创建 Topic 相关请参考 [数据订阅文档](../../develop/tmq/#创建-topic)。
+
+#### 创建 Consumer
+
+<Tabs defaultValue="native">
 
-<Tabs>
 <TabItem value="native" label="原生连接">
 
-##### 创建 stmt
+`Consumer` 提供了 Python 连接器订阅 TMQ 数据的 API。创建 Consumer 语法为 `consumer = Consumer(configs)`，参数定义请参考 [数据订阅文档](../../develop/tmq/#创建消费者-consumer)。
 
-Python 连接器的 `Connection` 提供了 `statement` 方法用于创建参数绑定对象 stmt，该方法接收 sql 字符串作为参数，sql 字符串目前仅支持用 `?` 来代表绑定的参数。
+```python
+from taos.tmq import Consumer
 
+consumer = Consumer({"group.id": "local", "td.connect.ip": "127.0.0.1"})
 ```
-import taos
+</TabItem>
 
-conn = taos.connect()
-stmt = conn.statement("insert into log values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
-```
+<TabItem value="websocket" label="WebSocket 连接">
 
-##### 参数绑定
+除了原生的连接方式，Python 连接器还支持通过 websocket 订阅 TMQ 数据，使用 websocket 方式订阅 TMQ 数据需要安装 `taos-ws-py`。
 
-调用 `new_multi_binds` 函数创建 params 列表，用于参数绑定。
+taosws `Consumer` API 提供了基于 Websocket 订阅 TMQ 数据的 API。创建 Consumer 语法为 `consumer = Consumer(conf=configs)`，使用时需要指定 `td.connect.websocket.scheme` 参数值为 "ws"，参数定义请参考 [数据订阅文档](../../develop/tmq/#%E5%88%9B%E5%BB%BA%E6%B6%88%E8%B4%B9%E8%80%85-consumer)。
 
-```
-params = new_multi_binds(16)
-params[0].timestamp((1626861392589, 1626861392590, 1626861392591))
-params[1].bool((True, None, False))
-params[2].tinyint([-128, -128, None])  # -128 is tinyint null
-params[3].tinyint([0, 127, None])
-params[4].smallint([3, None, 2])
-params[5].int([3, 4, None])
-params[6].bigint([3, 4, None])
-params[7].tinyint_unsigned([3, 4, None])
-params[8].smallint_unsigned([3, 4, None])
-params[9].int_unsigned([3, 4, None])
-params[10].bigint_unsigned([3, 4, None])
-params[11].float([3, None, 1])
-params[12].double([3, None, 1.2])
-params[13].binary(["abc", "dddafadfadfadfadfa", None])
-params[14].nchar(["涛思数据", None, "a long string with 中文字符"])
-params[15].timestamp([None, None, 1626861392591])
+```python
+import taosws
+
+consumer = taosws.(conf={"group.id": "local", "td.connect.websocket.scheme": "ws"})
 ```
 
-调用 stmt 的 `bind_param` 以单行的方式设置 values 或 `bind_param_batch` 以多行的方式设置 values 方法绑定参数。
+</TabItem>
+</Tabs>
 
-```
-stmt.bind_param_batch(params)
-```
+#### 订阅 topics
 
-##### 执行 sql
+<Tabs defaultValue="native">
 
-调用 stmt 的 `execute` 方法执行 sql
+<TabItem value="native" label="原生连接">
 
-```
-stmt.execute()
+Consumer API 的 `subscribe` 方法用于订阅 topics，consumer 支持同时订阅多个 topic。
+
+```python
+consumer.subscribe(['topic1', 'topic2'])
 ```
 
-##### 关闭 stmt
+</TabItem>
+<TabItem value="websocket" label="WebSocket 连接">
 
-最后需要关闭 stmt。
+Consumer API 的 `subscribe` 方法用于订阅 topics，consumer 支持同时订阅多个 topic。
 
+```python
+consumer.subscribe(['topic1', 'topic2'])
 ```
-stmt.close()
+
+</TabItem>
+</Tabs>
+
+#### 消费数据
+
+<Tabs defaultValue="native">
+
+<TabItem value="native" label="原生连接">
+
+Consumer API 的 `poll` 方法用于消费数据，`poll` 方法接收一个 float 类型的超时时间，超时时间单位为秒（s），`poll` 方法在超时之前返回一条 Message 类型的数据或超时返回 `None`。消费者必须通过 Message 的 `error()` 方法校验返回数据的 error 信息。
+
+```python
+while True:
+    res = consumer.poll(1)
+    if not res:
+        continue
+    err = res.error()
+    if err is not None:
+        raise err
+    val = res.value()
+
+    for block in val:
+        print(block.fetchall())
 ```
 
-##### 示例代码
+</TabItem>
+<TabItem value="websocket" label="WebSocket 连接">
+
+Consumer API 的 `poll` 方法用于消费数据，`poll` 方法接收一个 float 类型的超时时间，超时时间单位为秒（s），`poll` 方法在超时之前返回一条 Message 类型的数据或超时返回 `None`。消费者必须通过 Message 的 `error()` 方法校验返回数据的 error 信息。
 
 ```python
-{{#include docs/examples/python/stmt_example.py}}
+while True:
+    res = consumer.poll(timeout=1.0)
+    if not res:
+        continue
+    err = res.error()
+    if err is not None:
+        raise err
+    for block in message:
+        for row in block:
+            print(row)
 ```
+
 </TabItem>
+</Tabs>
 
-<TabItem value="websocket" label="WebSocket 连接">
+#### 获取消费进度
 
-##### 创建 stmt
+<Tabs defaultValue="native">
 
-Python WebSocket 连接器的 `Connection` 提供了 `statement` 方法用于创建参数绑定对象 stmt，该方法接收 sql 字符串作为参数，sql 字符串目前仅支持用 `?` 来代表绑定的参数。
+<TabItem value="native" label="原生连接">
+
+Consumer API 的 `assignment` 方法用于获取 Consumer 订阅的所有 topic 的消费进度，返回结果类型为 TopicPartition 列表。
 
+```python
+assignments = consumer.assignment()
 ```
-import taosws
 
-conn = taosws.connect('taosws://localhost:6041/test')
-stmt = conn.statement()
+Consumer API 的 `seek` 方法用于重置 Consumer 的消费进度到指定位置，方法参数类型为 TopicPartition。
+
+```python
+tp = TopicPartition(topic='topic1', partition=0, offset=0)
+consumer.seek(tp)
 ```
 
-##### 解析 sql
+</TabItem>
+<TabItem value="websocket" label="WebSocket 连接">
 
-调用 stmt 的 `prepare` 方法来解析 insert 语句。
+Consumer API 的 `assignment` 方法用于获取 Consumer 订阅的所有 topic 的消费进度，返回结果类型为 TopicPartition 列表。
 
+```python
+assignments = consumer.assignment()
 ```
-stmt.prepare("insert into t1 values (?, ?, ?, ?)")
+
+Consumer API 的 `seek` 方法用于重置 Consumer 的消费进度到指定位置。
+
+```python
+consumer.seek(topic='topic1', partition=0, offset=0)
 ```
 
-##### 参数绑定
+</TabItem>
+</Tabs>
 
-调用 stmt 的 `bind_param` 方法绑定参数。
+#### 关闭订阅
 
-```
-stmt.bind_param([
-    taosws.millis_timestamps_to_column([1686844800000, 1686844801000, 1686844802000, 1686844803000]),
-    taosws.ints_to_column([1, 2, 3, 4]),
-    taosws.floats_to_column([1.1, 2.2, 3.3, 4.4]),
-    taosws.varchar_to_column(['a', 'b', 'c', 'd']),
-])
-```
+<Tabs defaultValue="native">
 
-调用 stmt 的 `add_batch` 方法，将参数加入批处理。
+<TabItem value="native" label="原生连接">
 
-```
-stmt.add_batch()
+消费结束后，应当取消订阅，并关闭 Consumer。
+
+```python
+consumer.unsubscribe()
+consumer.close()
 ```
 
-##### 执行 sql
+</TabItem>
+<TabItem value="websocket" label="WebSocket 连接">
 
-调用 stmt 的 `execute` 方法执行 sql
+消费结束后，应当取消订阅，并关闭 Consumer。
 
-```
-stmt.execute()
+```python
+consumer.unsubscribe()
+consumer.close()
 ```
 
-##### 关闭 stmt
+</TabItem>
+</Tabs>
 
-最后需要关闭 stmt。
+#### 完整示例
+
+<Tabs defaultValue="native">
 
+<TabItem value="native" label="原生连接">
+
+```python
+{{#include docs/examples/python/tmq_example.py}}
 ```
-stmt.close()
+
+```python
+{{#include docs/examples/python/tmq_assignment_example.py:taos_get_assignment_and_seek_demo}}
 ```
 
-##### 示例代码
+</TabItem>
+<TabItem value="websocket" label="WebSocket 连接">
 
 ```python
-{{#include docs/examples/python/stmt_websocket_example.py}}
+{{#include docs/examples/python/tmq_websocket_example.py}}
+```
+
+```python
+{{#include docs/examples/python/tmq_websocket_assgnment_example.py:taosws_get_assignment_and_seek_demo}}
 ```
+
 </TabItem>
 </Tabs>
 

docs/zh/12-taos-sql/04-stable.md

@@ -51,6 +51,11 @@ DESCRIBE [db_name.]stb_name;
 
 ### 获取超级表中所有子表的标签信息
 
+```
+SHOW TABLE TAGS FROM table_name [FROM db_name];
+SHOW TABLE TAGS FROM [db_name.]table_name;
+```
+
 ```
 taos> SHOW TABLE TAGS FROM st1;
              tbname             |     id      |         loc          |

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

@@ -101,6 +101,7 @@ SHOW GRANTS;
 
 ```sql
 SHOW INDEXES FROM tbl_name [FROM db_name];
+SHOW INDEXES FROM [db_name.]tbl_name;
 ```
 
 显示已创建的索引。
@@ -269,6 +270,7 @@ Query OK, 24 row(s) in set (0.002444s)
 
 ```sql
 SHOW TAGS FROM child_table_name [FROM db_name];
+SHOW TAGS FROM [db_name.]child_table_name;
 ```
 
 显示子表的标签信息。

docs/zh/12-taos-sql/27-index.md

@@ -43,6 +43,7 @@ DROP INDEX index_name;
 ````sql
 ```sql
 SHOW INDEXES FROM tbl_name [FROM db_name];
+SHOW INDEXES FROM [db_name.]tbl_name;
 ````
 
 显示在所指定的数据库或表上已创建的索引。

include/common/ttokendef.h

@@ -355,8 +355,6 @@
 #define TK_WAL                            337
 
 
-
-
 #define TK_NK_SPACE   600
 #define TK_NK_COMMENT 601
 #define TK_NK_ILLEGAL 602

include/libs/stream/tstream.h

@@ -346,6 +346,7 @@ struct SStreamTask {
   int32_t             refCnt;
   int64_t             checkpointingId;
   int32_t             checkpointAlignCnt;
+  int32_t             transferStateAlignCnt;
   struct SStreamMeta* pMeta;
   SSHashObj*          pNameMap;
 };
@@ -630,6 +631,8 @@ int32_t streamProcessCheckpointRsp(SStreamMeta* pMeta, SStreamTask* pTask, SStre
 
 int32_t streamTaskReleaseState(SStreamTask* pTask);
 int32_t streamTaskReloadState(SStreamTask* pTask);
+int32_t streamAlignTransferState(SStreamTask* pTask);
+
 
 #ifdef __cplusplus
 }

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -265,22 +265,6 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
 
-#if 0
-  if (pMgmt->pTfs) {
-    if (tfsDirExistAt(pMgmt->pTfs, path, (SDiskID){0})) {
-      terrno = TSDB_CODE_VND_DIR_ALREADY_EXIST;
-      dError("vgId:%d, failed to restore vnode since %s", req.vgId, terrstr());
-      return -1;
-    }
-  } else {
-    if (taosDirExist(path)) {
-      terrno = TSDB_CODE_VND_DIR_ALREADY_EXIST;
-      dError("vgId:%d, failed to restore vnode since %s", req.vgId, terrstr());
-      return -1;
-    }
-  }
-#endif
-
   if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
     tFreeSCreateVnodeReq(&req);
     dError("vgId:%d, failed to create vnode since %s", req.vgId, terrstr());

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

@@ -911,12 +911,18 @@ int32_t mndDropSmasByStb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *p
       if (pStream != NULL && pStream->smaId == pSma->uid) {
         if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
           mError("stream:%s, failed to drop task since %s", pStream->name, terrstr());
+          mndReleaseStream(pMnode, pStream);
           goto _OVER;
         }
+
         if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
+          mndReleaseStream(pMnode, pStream);
           goto _OVER;
         }
+
+        mndReleaseStream(pMnode, pStream);
       }
+
       if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
       if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
       if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;

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

@@ -229,11 +229,11 @@ int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con
 int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type);
 int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
 int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer, int8_t rollback);
-void    tdRSmaQTaskInfoGetFileName(int32_t vgId, int64_t suid, int8_t level, int64_t version, char *outputName);
-void    tdRSmaQTaskInfoGetFullName(int32_t vgId, int64_t suid, int8_t level, int64_t version, const char *path,
+void    tdRSmaQTaskInfoGetFileName(SVnode *pVnode, int64_t suid, int8_t level, int64_t version, char *outputName);
+void    tdRSmaQTaskInfoGetFullName(SVnode *pVnode, int64_t suid, int8_t level, int64_t version, STfs *pTfs,
                                    char *outputName);
-void    tdRSmaQTaskInfoGetFullPath(int32_t vgId, int8_t level, const char *path, char *outputName);
-void    tdRSmaQTaskInfoGetFullPathEx(int32_t vgId, tb_uid_t suid, int8_t level, const char *path, char *outputName);
+void    tdRSmaQTaskInfoGetFullPath(SVnode *pVnode, int8_t level, STfs *pTfs, char *outputName);
+void    tdRSmaQTaskInfoGetFullPathEx(SVnode *pVnode, tb_uid_t suid, int8_t level, STfs *pTfs, char *outputName);
 
 static FORCE_INLINE void tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   int32_t ref = T_REF_INC(pRSmaInfo);
@@ -244,9 +244,9 @@ static FORCE_INLINE void tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   smaTrace("vgId:%d, unref rsma info:%p, val:%d", SMA_VID(pSma), pRSmaInfo, ref);
 }
 
-void tdRSmaGetFileName(int32_t vgId, const char *pdname, const char *dname, const char *fname, int64_t suid,
-                       int8_t level, int64_t version, char *outputName);
-void tdRSmaGetDirName(int32_t vgId, const char *pdname, const char *dname, bool endWithSep, char *outputName);
+void tdRSmaGetFileName(SVnode *pVnode, STfs *pTfs, const char *fname, int64_t suid, int8_t level, int64_t version,
+                       char *outputName);
+void tdRSmaGetDirName(SVnode *pVnode, STfs *pTfs, bool endWithSep, char *outputName);
 
 #ifdef __cplusplus
 }

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

@@ -87,6 +87,9 @@ void    vnodeBufPoolReset(SVBufPool* pPool);
 void    vnodeBufPoolAddToFreeList(SVBufPool* pPool);
 int32_t vnodeBufPoolRecycle(SVBufPool* pPool);
 
+// vnodeOpen.c
+int32_t vnodeGetPrimaryDir(const char* relPath, STfs* pTfs, char* buf, size_t bufLen);
+
 // vnodeQuery.c
 int32_t vnodeQueryOpen(SVnode* pVnode);
 void    vnodeQueryPreClose(SVnode* pVnode);

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

@@ -248,7 +248,7 @@ int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg);
-int32_t tqProcessTaskTransferStateReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessTaskTransferStateReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessStreamTaskScanHistoryFinishReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRecoverFinishRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqCheckLogInWal(STQ* pTq, int64_t version);

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

@@ -14,6 +14,7 @@
  */
 
 #include "meta.h"
+#include "vnd.h"
 
 static int tbDbKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
 static int skmDbKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
@@ -34,30 +35,27 @@ static void metaCleanup(SMeta **ppMeta);
 int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
   SMeta *pMeta = NULL;
   int    ret;
-  int    slen;
+  int    offset;
+  char   path[TSDB_FILENAME_LEN] = {0};
 
   *ppMeta = NULL;
 
   // create handle
-  if (pVnode->pTfs) {
-    slen = strlen(tfsGetPrimaryPath(pVnode->pTfs)) + strlen(pVnode->path) + strlen(VNODE_META_DIR) + 3;
-  } else {
-    slen = strlen(pVnode->path) + strlen(VNODE_META_DIR) + 2;
-  }
-  if ((pMeta = taosMemoryCalloc(1, sizeof(*pMeta) + slen)) == NULL) {
+  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, path, TSDB_FILENAME_LEN);
+  offset = strlen(path);
+  snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VNODE_META_DIR);
+
+  if ((pMeta = taosMemoryCalloc(1, sizeof(*pMeta) + strlen(path) + 1)) == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
 
   metaInitLock(pMeta);
+
   pMeta->path = (char *)&pMeta[1];
-  if (pVnode->pTfs) {
-    sprintf(pMeta->path, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP,
-            VNODE_META_DIR);
-  } else {
-    sprintf(pMeta->path, "%s%s%s", pVnode->path, TD_DIRSEP, VNODE_META_DIR);
-  }
-  taosRealPath(pMeta->path, NULL, slen);
+  strcpy(pMeta->path, path);
+  taosRealPath(pMeta->path, NULL, strlen(path) + 1);
+
   pMeta->pVnode = pVnode;
 
   // create path if not created yet

source/dnode/vnode/src/sma/smaFS.c

@@ -14,6 +14,7 @@
  */
 
 #include "sma.h"
+#include "vnd.h"
 
 // =================================================================================================
 
@@ -157,25 +158,15 @@ _exit:
 
 static void tdRSmaGetCurrentFName(SSma *pSma, char *current, char *current_t) {
   SVnode *pVnode = pSma->pVnode;
-  if (pVnode->pTfs) {
-    if (current) {
-      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%svnode%svnode%d%srsma%sPRESENT", tfsGetPrimaryPath(pVnode->pTfs),
-               TD_DIRSEP, TD_DIRSEP, TD_VID(pVnode), TD_DIRSEP, TD_DIRSEP);
-    }
-    if (current_t) {
-      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%svnode%svnode%d%srsma%sPRESENT.t", tfsGetPrimaryPath(pVnode->pTfs),
-               TD_DIRSEP, TD_DIRSEP, TD_VID(pVnode), TD_DIRSEP, TD_DIRSEP);
-    }
-  } else {
-#if 0
-    if (current) {
-      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%sPRESENT", pTsdb->path, TD_DIRSEP);
-    }
-    if (current_t) {
-      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%sPRESENT.t", pTsdb->path, TD_DIRSEP);
-    }
-#endif
-  }
+  int32_t offset = 0;
+
+  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, current, TSDB_FILENAME_LEN);
+  offset = strlen(current);
+  snprintf(current + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s%sPRESENT", TD_DIRSEP, VNODE_RSMA_DIR, TD_DIRSEP);
+
+  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, current_t, TSDB_FILENAME_LEN);
+  offset = strlen(current_t);
+  snprintf(current_t + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s%sPRESENT.t", TD_DIRSEP, VNODE_RSMA_DIR, TD_DIRSEP);
 }
 
 static int32_t tdRSmaLoadFSFromFile(const char *fname, SRSmaFS *pFS) {
@@ -309,8 +300,7 @@ static int32_t tdRSmaFSApplyChange(SSma *pSma, SRSmaFS *pFSNew) {
 
       nRef = atomic_sub_fetch_32(&preTaskF->nRef, 1);
       if (nRef <= 0) {
-        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), preTaskF->suid, preTaskF->level, preTaskF->version,
-                                   tfsGetPrimaryPath(pVnode->pTfs), fname);
+        tdRSmaQTaskInfoGetFullName(pVnode, preTaskF->suid, preTaskF->level, preTaskF->version, pVnode->pTfs, fname);
         (void)taosRemoveFile(fname);
         taosArrayRemove(pFSOld->aQTaskInf, idx);
       }
@@ -341,9 +331,9 @@ static int32_t tdRSmaFSScanAndTryFix(SSma *pSma) {
     SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, i);
 
     // main.tdb =========
-    tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, pTaskF->version,
-                               tfsGetPrimaryPath(pVnode->pTfs), fnameVer);
-    tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, -1, tfsGetPrimaryPath(pVnode->pTfs), fname);
+    tdRSmaQTaskInfoGetFullName(pVnode, pTaskF->suid, pTaskF->level, pTaskF->version,
+                               pVnode->pTfs, fnameVer);
+    tdRSmaQTaskInfoGetFullName(pVnode, pTaskF->suid, pTaskF->level, -1, pVnode->pTfs, fname);
 
     if (taosCheckExistFile(fnameVer)) {
       if (taosRenameFile(fnameVer, fname) < 0) {
@@ -597,8 +587,7 @@ void tdRSmaFSUnRef(SSma *pSma, SRSmaFS *pFS) {
 
     nRef = atomic_sub_fetch_32(&pTaskF->nRef, 1);
     if (nRef == 0) {
-      tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, pTaskF->version,
-                                 tfsGetPrimaryPath(pVnode->pTfs), fname);
+      tdRSmaQTaskInfoGetFullName(pVnode, pTaskF->suid, pTaskF->level, pTaskF->version, pVnode->pTfs, fname);
       if (taosRemoveFile(fname) < 0) {
         smaWarn("vgId:%d, failed to remove %s since %s", TD_VID(pVnode), fname, tstrerror(TAOS_SYSTEM_ERROR(errno)));
       } else {

source/dnode/vnode/src/sma/smaRollup.c

@@ -260,7 +260,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
     void       *pStreamState = NULL;
 
     // set the backend of stream state
-    tdRSmaQTaskInfoGetFullPathEx(TD_VID(pVnode), pRSmaInfo->suid, idx + 1, tfsGetPrimaryPath(pVnode->pTfs), taskInfDir);
+    tdRSmaQTaskInfoGetFullPathEx(pVnode, pRSmaInfo->suid, idx + 1, pVnode->pTfs, taskInfDir);
     if (!taosCheckExistFile(taskInfDir)) {
       char *s = taosStrdup(taskInfDir);
       if (taosMulMkDir(taosDirName(s)) != 0) {
@@ -1258,9 +1258,8 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
                  pRSmaInfo->suid, i + 1);
 
         // qTaskInfo file
-        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pRSmaInfo->suid, i + 1, -1, tfsGetPrimaryPath(pVnode->pTfs), fname);
-        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pRSmaInfo->suid, i + 1, version, tfsGetPrimaryPath(pVnode->pTfs),
-                                   fnameVer);
+        tdRSmaQTaskInfoGetFullName(pVnode, pRSmaInfo->suid, i + 1, -1, pVnode->pTfs, fname);
+        tdRSmaQTaskInfoGetFullName(pVnode, pRSmaInfo->suid, i + 1, version, pVnode->pTfs, fnameVer);
         if (taosCheckExistFile(fnameVer)) {
           smaWarn("vgId:%d, rsma persist, duplicate file %s exist", TD_VID(pVnode), fnameVer);
         }

source/dnode/vnode/src/sma/smaSnapshot.c

@@ -117,8 +117,7 @@ static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppBuf)
       continue;
     }
 
-    tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), qTaskF->suid, qTaskF->level, version, tfsGetPrimaryPath(pVnode->pTfs),
-                               fname);
+    tdRSmaQTaskInfoGetFullName(pVnode, qTaskF->suid, qTaskF->level, version, pVnode->pTfs, fname);
     if (!taosCheckExistFile(fname)) {
       smaError("vgId:%d, vnode snapshot rsma reader for qtaskinfo, table %" PRIi64 ", level %" PRIi8
                ", version %" PRIi64 " failed since %s not exist",
@@ -340,7 +339,6 @@ int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback) {
   SSmaEnv*         pEnv = NULL;
   SRSmaStat*       pStat = NULL;
   SRSmaSnapWriter* pWriter = *ppWriter;
-  const char*      primaryPath = NULL;
   char             fname[TSDB_FILENAME_LEN] = {0};
   char             fnameVer[TSDB_FILENAME_LEN] = {0};
   TdFilePtr        pOutFD = NULL;
@@ -354,7 +352,6 @@ int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback) {
   pVnode = pSma->pVnode;
   pEnv = SMA_RSMA_ENV(pSma);
   pStat = (SRSmaStat*)SMA_ENV_STAT(pEnv);
-  primaryPath = tfsGetPrimaryPath(pVnode->pTfs);
 
   // rsma1/rsma2
   for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
@@ -375,8 +372,8 @@ int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback) {
     for (int32_t i = 0; i < size; ++i) {
       SQTaskFile* pTaskF = TARRAY_GET_ELEM(pFS->aQTaskInf, i);
       if (pTaskF->version == pWriter->ever) {
-        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, pTaskF->version, primaryPath, fnameVer);
-        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, -1, primaryPath, fname);
+        tdRSmaQTaskInfoGetFullName(pVnode, pTaskF->suid, pTaskF->level, pTaskF->version, pVnode->pTfs, fnameVer);
+        tdRSmaQTaskInfoGetFullName(pVnode, pTaskF->suid, pTaskF->level, -1, pVnode->pTfs, fname);
 
         pInFD = taosOpenFile(fnameVer, TD_FILE_READ);
         if (pInFD == NULL) {
@@ -486,8 +483,7 @@ static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData,
   SQTaskFile qTaskFile = {
       .nRef = 1, .level = pHdr->flag, .suid = pHdr->index, .version = pWriter->ever, .size = pHdr->size};
 
-  tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pHdr->index, pHdr->flag, qTaskFile.version,
-                             tfsGetPrimaryPath(pVnode->pTfs), fname);
+  tdRSmaQTaskInfoGetFullName(pVnode, pHdr->index, pHdr->flag, qTaskFile.version, pVnode->pTfs, fname);
 
   fp = taosCreateFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
   if (!fp) {

source/dnode/vnode/src/sma/smaUtil.c

@@ -14,88 +14,72 @@
  */
 
 #include "sma.h"
+#include "vnd.h"
 
 #define TD_QTASKINFO_FNAME_PREFIX "main.tdb"
 
-void tdRSmaQTaskInfoGetFileName(int32_t vgId, int64_t suid, int8_t level, int64_t version, char *outputName) {
-  tdRSmaGetFileName(vgId, NULL, VNODE_RSMA_DIR, TD_QTASKINFO_FNAME_PREFIX, suid, level, version, outputName);
+void tdRSmaQTaskInfoGetFileName(SVnode *pVnode, int64_t suid, int8_t level, int64_t version, char *outputName) {
+  tdRSmaGetFileName(pVnode, NULL, TD_QTASKINFO_FNAME_PREFIX, suid, level, version, outputName);
 }
 
-void tdRSmaQTaskInfoGetFullName(int32_t vgId, int64_t suid, int8_t level, int64_t version, const char *path,
+void tdRSmaQTaskInfoGetFullName(SVnode *pVnode, int64_t suid, int8_t level, int64_t version, STfs *pTfs,
                                 char *outputName) {
-  tdRSmaGetFileName(vgId, path, VNODE_RSMA_DIR, TD_QTASKINFO_FNAME_PREFIX, suid, level, version, outputName);
+  tdRSmaGetFileName(pVnode, pTfs, TD_QTASKINFO_FNAME_PREFIX, suid, level, version, outputName);
 }
 
-void tdRSmaQTaskInfoGetFullPath(int32_t vgId, int8_t level, const char *path, char *outputName) {
-  tdRSmaGetDirName(vgId, path, VNODE_RSMA_DIR, true, outputName);
+void tdRSmaQTaskInfoGetFullPath(SVnode *pVnode, int8_t level, STfs *pTfs, char *outputName) {
+  tdRSmaGetDirName(pVnode, pTfs, true, outputName);
   int32_t rsmaLen = strlen(outputName);
   snprintf(outputName + rsmaLen, TSDB_FILENAME_LEN - rsmaLen, "%" PRIi8, level);
 }
 
-void tdRSmaQTaskInfoGetFullPathEx(int32_t vgId, tb_uid_t suid, int8_t level, const char *path, char *outputName) {
-  tdRSmaGetDirName(vgId, path, VNODE_RSMA_DIR, true, outputName);
+void tdRSmaQTaskInfoGetFullPathEx(SVnode *pVnode, tb_uid_t suid, int8_t level, STfs *pTfs, char *outputName) {
+  tdRSmaGetDirName(pVnode, pTfs, true, outputName);
   int32_t rsmaLen = strlen(outputName);
   snprintf(outputName + rsmaLen, TSDB_FILENAME_LEN - rsmaLen, "%" PRIi8 "%s%" PRIi64, level, TD_DIRSEP, suid);
 }
 
-void tdRSmaGetFileName(int32_t vgId, const char *pdname, const char *dname, const char *fname, int64_t suid,
-                       int8_t level, int64_t version, char *outputName) {
+void tdRSmaGetFileName(SVnode *pVnode, STfs *pTfs, const char *fname, int64_t suid, int8_t level, int64_t version,
+                       char *outputName) {
+  int32_t offset = 0;
+
+  // vnode
+  vnodeGetPrimaryDir(pVnode->path, pTfs, outputName, TSDB_FILENAME_LEN);
+  offset = strlen(outputName);
+
+  // rsma
+  snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VNODE_RSMA_DIR);
+  offset = strlen(outputName);
+
+  // level & suid || vgid
   if (level >= 0 && suid > 0) {
-    if (version >= 0) {
-      if (pdname) {
-        snprintf(outputName, TSDB_FILENAME_LEN, "%s%svnode%svnode%d%s%s%s%" PRIi8 "%s%" PRIi64 "%s%s.%" PRIi64, pdname,
-                 TD_DIRSEP, TD_DIRSEP, vgId, TD_DIRSEP, dname, TD_DIRSEP, level, TD_DIRSEP, suid, TD_DIRSEP, fname,
-                 version);
+    snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%" PRIi8 "%s%" PRIi64 "%s", TD_DIRSEP, level,
+             TD_DIRSEP, suid, TD_DIRSEP);
   } else {
-        snprintf(outputName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s%s%" PRIi8 "%s%" PRIi64 "%s%s.%" PRIi64, TD_DIRSEP,
-                 vgId, TD_DIRSEP, dname, TD_DIRSEP, level, TD_DIRSEP, suid, TD_DIRSEP, fname, version);
+    snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, "%sv%d", TD_DIRSEP, TD_VID(pVnode));
   }
-    } else {
-      if (pdname) {
-        snprintf(outputName, TSDB_FILENAME_LEN, "%s%svnode%svnode%d%s%s%s%" PRIi8 "%s%" PRIi64 "%s%s", pdname,
-                 TD_DIRSEP, TD_DIRSEP, vgId, TD_DIRSEP, dname, TD_DIRSEP, level, TD_DIRSEP, suid, TD_DIRSEP, fname);
-      } else {
-        snprintf(outputName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s%s%" PRIi8 "%s%" PRIi64 "%s%s", TD_DIRSEP, vgId,
-                 TD_DIRSEP, dname, TD_DIRSEP, level, TD_DIRSEP, suid, TD_DIRSEP, fname);
-      }
-    }
-  } else {
+  offset = strlen(outputName);
+
+  // fname
+  snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, "%s", fname);
+  offset = strlen(outputName);
+
+  // version
   if (version >= 0) {
-      if (pdname) {
-        snprintf(outputName, TSDB_FILENAME_LEN, "%s%svnode%svnode%d%s%s%sv%d%s%" PRIi64, pdname, TD_DIRSEP, TD_DIRSEP,
-                 vgId, TD_DIRSEP, dname, TD_DIRSEP, vgId, fname, version);
-      } else {
-        snprintf(outputName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s%sv%d%s%" PRIi64, TD_DIRSEP, vgId, TD_DIRSEP, dname,
-                 TD_DIRSEP, vgId, fname, version);
-      }
-    } else {
-      if (pdname) {
-        snprintf(outputName, TSDB_FILENAME_LEN, "%s%svnode%svnode%d%s%s%sv%d%s", pdname, TD_DIRSEP, TD_DIRSEP, vgId,
-                 TD_DIRSEP, dname, TD_DIRSEP, vgId, fname);
-      } else {
-        snprintf(outputName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s%sv%d%s", TD_DIRSEP, vgId, TD_DIRSEP, dname,
-                 TD_DIRSEP, vgId, fname);
-      }
-    }
+    snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, ".%" PRIi64, version);
   }
 }
 
-void tdRSmaGetDirName(int32_t vgId, const char *pdname, const char *dname, bool endWithSep, char *outputName) {
-  if (pdname) {
-    if (endWithSep) {
-      snprintf(outputName, TSDB_FILENAME_LEN, "%s%svnode%svnode%d%s%s%s", pdname, TD_DIRSEP, TD_DIRSEP, vgId, TD_DIRSEP,
-               dname, TD_DIRSEP);
-    } else {
-      snprintf(outputName, TSDB_FILENAME_LEN, "%s%svnode%svnode%d%s%s", pdname, TD_DIRSEP, TD_DIRSEP, vgId, TD_DIRSEP,
-               dname);
-    }
-  } else {
-    if (endWithSep) {
-      snprintf(outputName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s%s", TD_DIRSEP, vgId, TD_DIRSEP, dname, TD_DIRSEP);
-    } else {
-      snprintf(outputName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s", TD_DIRSEP, vgId, TD_DIRSEP, dname);
-    }
-  }
+void tdRSmaGetDirName(SVnode *pVnode, STfs *pTfs, bool endWithSep, char *outputName) {
+  int32_t offset = 0;
+
+  // vnode
+  vnodeGetPrimaryDir(pVnode->path, pTfs, outputName, TSDB_FILENAME_LEN);
+  offset = strlen(outputName);
+
+  // rsma
+  snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s%s", TD_DIRSEP, VNODE_RSMA_DIR,
+           (endWithSep ? TD_DIRSEP : ""));
 }
 
 // smaXXXUtil ================

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

@@ -934,7 +934,6 @@ int32_t tqProcessStreamTaskCheckReq(STQ* pTq, SRpcMsg* pMsg) {
   };
 
   SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, taskId);
-
   if (pTask != NULL) {
     rsp.status = streamTaskCheckStatus(pTask);
     streamMetaReleaseTask(pTq->pStreamMeta, pTask);
@@ -1106,7 +1105,15 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
       // 1. stop the related stream task, get the current scan wal version of stream task, ver.
       pStreamTask = streamMetaAcquireTask(pMeta, pTask->streamTaskId.taskId);
       if (pStreamTask == NULL) {
-        // todo handle error
+        qError("failed to find s-task:0x%x, it may have been destroyed, drop fill history task:%s",
+               pTask->streamTaskId.taskId, pTask->id.idStr);
+
+        pTask->status.taskStatus = TASK_STATUS__DROPPING;
+        tqDebug("s-task:%s scan-history-task set status to be dropping", pId);
+
+        streamMetaSaveTask(pMeta, pTask);
+        streamMetaReleaseTask(pMeta, pTask);
+        return -1;
       }
 
       ASSERT(pStreamTask->info.taskLevel == TASK_LEVEL__SOURCE);
@@ -1213,11 +1220,14 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
 }
 
 // notify the downstream tasks to transfer executor state after handle all history blocks.
-int32_t tqProcessTaskTransferStateReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
-  SStreamTransferReq req;
+int32_t tqProcessTaskTransferStateReq(STQ* pTq, SRpcMsg* pMsg) {
+  char*   pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
+  int32_t len = pMsg->contLen - sizeof(SMsgHead);
+
+  SStreamTransferReq req = {0};
 
   SDecoder decoder;
-  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
+  tDecoderInit(&decoder, (uint8_t*)pReq, len);
   int32_t code = tDecodeStreamScanHistoryFinishReq(&decoder, &req);
   tDecoderClear(&decoder);
 
@@ -1227,25 +1237,33 @@ int32_t tqProcessTaskTransferStateReq(STQ* pTq, int64_t sversion, char* msg, int
     return -1;
   }
 
+  int32_t remain = streamAlignTransferState(pTask);
+  if (remain > 0) {
+    tqDebug("s-task:%s receive transfer state msg, remain:%d", pTask->id.idStr, remain);
+    return 0;
+  }
+
   // transfer the ownership of executor state
-  streamTaskReleaseState(pTask);
-  tqDebug("s-task:%s receive state transfer req", pTask->id.idStr);
+  tqDebug("s-task:%s all upstream tasks end transfer msg", pTask->id.idStr);
 
   // related stream task load the state from the state storage backend
   SStreamTask* pStreamTask = streamMetaAcquireTask(pTq->pStreamMeta, pTask->streamTaskId.taskId);
   if (pStreamTask == NULL) {
+    streamMetaReleaseTask(pTq->pStreamMeta, pTask);
     tqError("failed to find related stream task:0x%x, it may have been dropped already", req.taskId);
     return -1;
   }
 
+  // when all upstream tasks have notified the this task to start transfer state, then we start the transfer procedure.
+  streamTaskReleaseState(pTask);
   streamTaskReloadState(pStreamTask);
+  streamMetaReleaseTask(pTq->pStreamMeta, pStreamTask);
 
   ASSERT(pTask->streamTaskId.taskId != 0);
   pTask->status.transferState = true;
 
   streamSchedExec(pTask);
   streamMetaReleaseTask(pTq->pStreamMeta, pTask);
-
   return 0;
 }
 

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

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 #include "tsdb.h"
+#include "vnd.h"
 
 #define ROCKS_BATCH_SIZE (4096)
 
@@ -58,16 +59,10 @@ typedef struct {
 
 static void tsdbGetRocksPath(STsdb *pTsdb, char *path) {
   SVnode *pVnode = pTsdb->pVnode;
-  if (pVnode->pTfs) {
-    if (path) {
-      snprintf(path, TSDB_FILENAME_LEN, "%s%s%s%scache.rdb", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP,
-               pTsdb->path, TD_DIRSEP);
-    }
-  } else {
-    if (path) {
-      snprintf(path, TSDB_FILENAME_LEN, "%s%scache.rdb", pTsdb->path, TD_DIRSEP);
-    }
-  }
+  vnodeGetPrimaryDir(pTsdb->path, pVnode->pTfs, path, TSDB_FILENAME_LEN);
+
+  int32_t offset = strlen(path);
+  snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%scache.rdb", TD_DIRSEP);
 }
 
 static const char *myCmpName(void *state) {

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

@@ -14,6 +14,7 @@
  */
 
 #include "tsdb.h"
+#include "vnd.h"
 
 // =================================================================================================
 static int32_t tsdbFSToBinary(uint8_t *p, STsdbFS *pFS) {
@@ -271,22 +272,20 @@ int32_t tDFileSetCmprFn(const void *p1, const void *p2) {
 
 void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) {
   SVnode *pVnode = pTsdb->pVnode;
-  if (pVnode->pTfs) {
-    if (current) {
-      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%s%s%sCURRENT", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP,
-               pTsdb->path, TD_DIRSEP);
-    }
-    if (current_t) {
-      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%s%s%sCURRENT.t", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP,
-               pTsdb->path, TD_DIRSEP);
-    }
-  } else {
+  int32_t offset = 0;
+
+  // CURRENT
   if (current) {
-      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%sCURRENT", pTsdb->path, TD_DIRSEP);
+    vnodeGetPrimaryDir(pTsdb->path, pVnode->pTfs, current, TSDB_FILENAME_LEN);
+    offset = strlen(current);
+    snprintf(current + offset, TSDB_FILENAME_LEN - offset - 1, "%sCURRENT", TD_DIRSEP);
   }
+
+  // CURRENT.t
   if (current_t) {
-      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%sCURRENT.t", pTsdb->path, TD_DIRSEP);
-    }
+    vnodeGetPrimaryDir(pTsdb->path, pVnode->pTfs, current_t, TSDB_FILENAME_LEN);
+    offset = strlen(current_t);
+    snprintf(current_t + offset, TSDB_FILENAME_LEN - offset - 1, "%sCURRENT.t", TD_DIRSEP);
   }
 }
 

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

@@ -14,6 +14,7 @@
  */
 
 #include "tsdb.h"
+#include "vnd.h"
 
 int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile) {
   int32_t n = 0;
@@ -282,8 +283,12 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
 
 // SDelFile ===============================================
 void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]) {
-  snprintf(fname, TSDB_FILENAME_LEN - 1, "%s%s%s%sv%dver%" PRId64 "%s", tfsGetPrimaryPath(pTsdb->pVnode->pTfs),
-           TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), pFile->commitID, ".del");
+  int32_t offset = 0;
+
+  vnodeGetPrimaryDir(pTsdb->path, pTsdb->pVnode->pTfs, fname, TSDB_FILENAME_LEN);
+  offset = strlen(fname);
+  snprintf((char *)fname + offset, TSDB_FILENAME_LEN - offset - 1, "%sv%dver%" PRId64 ".del", TD_DIRSEP,
+           TD_VID(pTsdb->pVnode), pFile->commitID);
 }
 
 int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile) {

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

@@ -295,11 +295,7 @@ static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) {
   pInfo->txn = metaGetTxn(pVnode->pMeta);
 
   // save info
-  if (pVnode->pTfs) {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
-  } else {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
-  }
+  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode));
   if (vnodeSaveInfo(dir, &pInfo->info) < 0) {
@@ -432,11 +428,7 @@ static int vnodeCommitImpl(SCommitInfo *pInfo) {
     return -1;
   }
 
-  if (pVnode->pTfs) {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
-  } else {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
-  }
+  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed);
 
@@ -497,16 +489,22 @@ _exit:
 
 bool vnodeShouldRollback(SVnode *pVnode) {
   char tFName[TSDB_FILENAME_LEN] = {0};
-  snprintf(tFName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP,
-           VND_INFO_FNAME_TMP);
+  int32_t offset = 0;
+
+  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
+  offset = strlen(tFName);
+  snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
 
   return taosCheckExistFile(tFName);
 }
 
 void vnodeRollback(SVnode *pVnode) {
   char tFName[TSDB_FILENAME_LEN] = {0};
-  snprintf(tFName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP,
-           VND_INFO_FNAME_TMP);
+  int32_t offset = 0;
+
+  vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
+  offset = strlen(tFName);
+  snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
 
   (void)taosRemoveFile(tFName);
 }

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

@@ -15,6 +15,16 @@
 
 #include "vnd.h"
 
+int32_t vnodeGetPrimaryDir(const char *relPath, STfs *pTfs, char *buf, size_t bufLen) {
+  if (pTfs) {
+    snprintf(buf, bufLen - 1, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, relPath);
+  } else {
+    snprintf(buf, bufLen - 1, "%s", relPath);
+  }
+  buf[bufLen - 1] = '\0';
+  return 0;
+}
+
 int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
@@ -26,18 +36,10 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
   }
 
   // create vnode env
-  if (pTfs) {
-    if (tfsMkdirAt(pTfs, path, (SDiskID){0}) < 0) {
-      vError("vgId:%d, failed to create vnode since:%s", pCfg->vgId, tstrerror(terrno));
-      return -1;
-    }
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, path);
-  } else {
-    if (taosMkDir(path)) {
+  vnodeGetPrimaryDir(path, pTfs, dir, TSDB_FILENAME_LEN);
+  if (taosMkDir(dir)) {
     return TAOS_SYSTEM_ERROR(errno);
   }
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", path);
-  }
 
   if (pCfg) {
     info.config = *pCfg;
@@ -63,11 +65,7 @@ int32_t vnodeAlterReplica(const char *path, SAlterVnodeReplicaReq *pReq, STfs *p
   char       dir[TSDB_FILENAME_LEN] = {0};
   int32_t    ret = 0;
 
-  if (pTfs) {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, path);
-  } else {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", path);
-  }
+  vnodeGetPrimaryDir(path, pTfs, dir, TSDB_FILENAME_LEN);
 
   ret = vnodeLoadInfo(dir, &info);
   if (ret < 0) {
@@ -185,21 +183,12 @@ int32_t vnodeRenameVgroupId(const char *srcPath, const char *dstPath, int32_t sr
   return ret;
 }
 
-int32_t vnodeGetAbsDir(const char *relPath, STfs *pTfs, char *buf, size_t bufLen) {
-  if (pTfs) {
-    snprintf(buf, bufLen, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, relPath);
-  } else {
-    snprintf(buf, bufLen, "%s", relPath);
-  }
-  return 0;
-}
-
 int32_t vnodeAlterHashRange(const char *srcPath, const char *dstPath, SAlterVnodeHashRangeReq *pReq, STfs *pTfs) {
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
   int32_t    ret = 0;
 
-  vnodeGetAbsDir(srcPath, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(srcPath, pTfs, dir, TSDB_FILENAME_LEN);
 
   ret = vnodeLoadInfo(dir, &info);
   if (ret < 0) {
@@ -258,7 +247,7 @@ int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t s
   SVnodeInfo info = {0};
   char       dir[TSDB_FILENAME_LEN] = {0};
 
-  vnodeGetAbsDir(dstPath, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(dstPath, pTfs, dir, TSDB_FILENAME_LEN);
   if (vnodeLoadInfo(dir, &info) == 0) {
     if (info.config.vgId != dstVgId) {
       vError("vgId:%d, unexpected vnode config.vgId:%d", dstVgId, info.config.vgId);
@@ -267,7 +256,7 @@ int32_t vnodeRestoreVgroupId(const char *srcPath, const char *dstPath, int32_t s
     return dstVgId;
   }
 
-  vnodeGetAbsDir(srcPath, pTfs, dir, TSDB_FILENAME_LEN);
+  vnodeGetPrimaryDir(srcPath, pTfs, dir, TSDB_FILENAME_LEN);
   if (vnodeLoadInfo(dir, &info) < 0) {
     vError("vgId:%d, failed to read vnode config from %s since %s", srcVgId, srcPath, tstrerror(terrno));
     return -1;
@@ -302,11 +291,7 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
   char       tdir[TSDB_FILENAME_LEN * 2] = {0};
   int32_t    ret = 0;
 
-  if (pTfs) {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, path);
-  } else {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", path);
-  }
+  vnodeGetPrimaryDir(path, pTfs, dir, TSDB_FILENAME_LEN);
 
   info.config = vnodeCfgDefault;
 

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

@@ -91,12 +91,11 @@ int32_t vnodeSnapRead(SVSnapReader *pReader, uint8_t **ppData, uint32_t *nData)
   // FIXME: if commit multiple times and the config changed?
   if (!pReader->cfgDone) {
     char fName[TSDB_FILENAME_LEN];
-    if (pReader->pVnode->pTfs) {
-      snprintf(fName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pReader->pVnode->pTfs), TD_DIRSEP,
-               pReader->pVnode->path, TD_DIRSEP, VND_INFO_FNAME);
-    } else {
-      snprintf(fName, TSDB_FILENAME_LEN, "%s%s%s", pReader->pVnode->path, TD_DIRSEP, VND_INFO_FNAME);
-    }
+    int32_t offset = 0;
+
+    vnodeGetPrimaryDir(pReader->pVnode->path, pReader->pVnode->pTfs, fName, TSDB_FILENAME_LEN);
+    offset = strlen(fName);
+    snprintf(fName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME);
 
     TdFilePtr pFile = taosOpenFile(fName, TD_FILE_READ);
     if (NULL == pFile) {
@@ -344,11 +343,7 @@ int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback, SSnapshot *
                               .applyTerm = pWriter->info.state.commitTerm};
     pVnode->statis = pWriter->info.statis;
     char dir[TSDB_FILENAME_LEN] = {0};
-    if (pWriter->pVnode->pTfs) {
-      snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
-    } else {
-      snprintf(dir, TSDB_FILENAME_LEN, "%s", pWriter->pVnode->path);
-    }
+    vnodeGetPrimaryDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
     vnodeCommitInfo(dir);
   } else {
@@ -400,12 +395,7 @@ static int32_t vnodeSnapWriteInfo(SVSnapWriter *pWriter, uint8_t *pData, uint32_
 
   // modify info as needed
   char dir[TSDB_FILENAME_LEN] = {0};
-  if (pWriter->pVnode->pTfs) {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pWriter->pVnode->pTfs), TD_DIRSEP,
-             pWriter->pVnode->path);
-  } else {
-    snprintf(dir, TSDB_FILENAME_LEN, "%s", pWriter->pVnode->path);
-  }
+  vnodeGetPrimaryDir(pWriter->pVnode->path, pWriter->pVnode->pTfs, dir, TSDB_FILENAME_LEN);
 
   SVnodeStats vndStats = pWriter->info.config.vndStats;
   SVnode     *pVnode = pWriter->pVnode;

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

@@ -661,11 +661,8 @@ int32_t vnodeProcessStreamMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo)
       return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);
     case TDMT_VND_STREAM_SCAN_HISTORY:
       return tqProcessTaskScanHistory(pVnode->pTq, pMsg);
-    case TDMT_STREAM_TRANSFER_STATE: {
-      char* pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
-      int32_t len = pMsg->contLen - sizeof(SMsgHead);
-      return tqProcessTaskTransferStateReq(pVnode->pTq, 0, pReq, len);
-    }
+    case TDMT_STREAM_TRANSFER_STATE:
+      return tqProcessTaskTransferStateReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_SCAN_HISTORY_FINISH:
       return tqProcessStreamTaskScanHistoryFinishReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_SCAN_HISTORY_FINISH_RSP:

source/libs/command/src/command.c

@@ -286,6 +286,9 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char* dbName, ch
     hashPrefix = pCfg->hashPrefix + dbFNameLen + 1;
   }
 
+  if (IS_SYS_DBNAME(dbName)) {
+    len += sprintf(buf2 + VARSTR_HEADER_SIZE, "CREATE DATABASE `%s`", dbName);
+  } else {
     len += sprintf(
         buf2 + VARSTR_HEADER_SIZE,
         "CREATE DATABASE `%s` BUFFER %d CACHESIZE %d CACHEMODEL '%s' COMP %d DURATION %dm "
@@ -301,6 +304,7 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char* dbName, ch
       len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, " RETENTIONS %s", retentions);
       taosMemoryFree(retentions);
     }
+  }
 
   (varDataLen(buf2)) = len;
 

source/libs/executor/src/executor.c

@@ -288,8 +288,7 @@ qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int3
     return pTaskInfo;
   }
 
-  struct SSubplan* pPlan = NULL;
-
+  SSubplan* pPlan = NULL;
   int32_t   code = qStringToSubplan(msg, &pPlan);
   if (code != TSDB_CODE_SUCCESS) {
     terrno = code;
@@ -335,6 +334,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers, int32_t v
   qTaskInfo_t pTaskInfo = NULL;
   code = qCreateExecTask(readers, vgId, 0, pPlan, &pTaskInfo, NULL, NULL, OPTR_EXEC_MODEL_STREAM);
   if (code != TSDB_CODE_SUCCESS) {
+    nodesDestroyNode((SNode*)pPlan);
     qDestroyTask(pTaskInfo);
     terrno = code;
     return NULL;

source/libs/parser/inc/sql.y

@@ -447,6 +447,7 @@ cmd ::= SHOW MNODES.
 cmd ::= SHOW QNODES.                                                              { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_QNODES_STMT); }
 cmd ::= SHOW FUNCTIONS.                                                           { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_FUNCTIONS_STMT); }
 cmd ::= SHOW INDEXES FROM table_name_cond(A) from_db_opt(B).                      { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_INDEXES_STMT, B, A, OP_TYPE_EQUAL); }
+cmd ::= SHOW INDEXES FROM db_name(B) NK_DOT table_name(A).                        { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_INDEXES_STMT, createIdentifierValueNode(pCxt, &B), createIdentifierValueNode(pCxt, &A), OP_TYPE_EQUAL); }
 cmd ::= SHOW STREAMS.                                                             { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_STREAMS_STMT); }
 cmd ::= SHOW ACCOUNTS.                                                            { pCxt->errCode = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_EXPRIE_STATEMENT); }
 cmd ::= SHOW APPS.                                                                { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_APPS_STMT); }
@@ -471,7 +472,9 @@ cmd ::= SHOW TABLE DISTRIBUTED full_table_name(A).
 cmd ::= SHOW CONSUMERS.                                                           { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_CONSUMERS_STMT); }
 cmd ::= SHOW SUBSCRIPTIONS.                                                       { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT); }
 cmd ::= SHOW TAGS FROM table_name_cond(A) from_db_opt(B).                         { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TAGS_STMT, B, A, OP_TYPE_EQUAL); }
+cmd ::= SHOW TAGS FROM db_name(B) NK_DOT table_name(A).                           { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TAGS_STMT, createIdentifierValueNode(pCxt, &B), createIdentifierValueNode(pCxt, &A), OP_TYPE_EQUAL); }
 cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM table_name_cond(A) from_db_opt(B).   { pCxt->pRootNode = createShowTableTagsStmt(pCxt, A, B, C); }
+cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM db_name(B) NK_DOT table_name(A).     { pCxt->pRootNode = createShowTableTagsStmt(pCxt, createIdentifierValueNode(pCxt, &A), createIdentifierValueNode(pCxt, &B), C); }
 cmd ::= SHOW VNODES NK_INTEGER(A).                                                { pCxt->pRootNode = createShowVnodesStmt(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &A), NULL); }
 cmd ::= SHOW VNODES NK_STRING(A).                                                 { pCxt->pRootNode = createShowVnodesStmt(pCxt, NULL, createValueNode(pCxt, TSDB_DATA_TYPE_VARCHAR, &A)); }
 // show alive

source/libs/parser/src/parTranslater.c

@@ -509,6 +509,10 @@ static int32_t getDBVgVersion(STranslateContext* pCxt, const char* pDbFName, int
 }
 
 static int32_t getDBCfg(STranslateContext* pCxt, const char* pDbName, SDbCfgInfo* pInfo) {
+  if (IS_SYS_DBNAME(pDbName)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
   SParseContext* pParCxt = pCxt->pParseCxt;
   SName          name;
   tNameSetDbName(&name, pCxt->pParseCxt->acctId, pDbName, strlen(pDbName));

source/libs/stream/src/stream.c

@@ -61,7 +61,7 @@ char* createStreamTaskIdStr(int64_t streamId, int32_t taskId) {
   return taosStrdup(buf);
 }
 
-void streamSchedByTimer(void* param, void* tmrId) {
+static void streamSchedByTimer(void* param, void* tmrId) {
   SStreamTask* pTask = (void*)param;
 
   int8_t status = atomic_load_8(&pTask->triggerStatus);

source/libs/stream/src/streamExec.c

@@ -352,11 +352,12 @@ static void waitForTaskIdle(SStreamTask* pTask, SStreamTask* pStreamTask) {
 static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
   SStreamTask* pStreamTask = streamMetaAcquireTask(pTask->pMeta, pTask->streamTaskId.taskId);
   if (pStreamTask == NULL) {
-    qError("s-task:%s failed to find related stream task:0x%x, it may have been destoryed or closed",
+    qError("s-task:%s failed to find related stream task:0x%x, it may have been destroyed or closed",
         pTask->id.idStr, pTask->streamTaskId.taskId);
     return TSDB_CODE_STREAM_TASK_NOT_EXIST;
   } else {
-    qDebug("s-task:%s scan history task end, update stream task:%s info, transfer exec state", pTask->id.idStr, pStreamTask->id.idStr);
+    qDebug("s-task:%s fill-history task end, update related stream task:%s info, transfer exec state", pTask->id.idStr,
+           pStreamTask->id.idStr);
   }
 
   ASSERT(pStreamTask != NULL && pStreamTask->historyTaskId.taskId == pTask->id.taskId);
@@ -369,6 +370,7 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
   } else {
     ASSERT(pStreamTask->status.taskStatus == TASK_STATUS__NORMAL);
     pStreamTask->status.taskStatus = TASK_STATUS__HALT;
+    qDebug("s-task:%s status: halt by related fill history task:%s", pStreamTask->id.idStr, pTask->id.idStr);
   }
 
   // wait for the stream task to be idle
@@ -477,6 +479,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
       ASSERT(batchSize == 0);
       if (pTask->info.fillHistory && pTask->status.transferState) {
         int32_t code = streamTransferStateToStreamTask(pTask);
+        pTask->status.transferState = false;  // reset this value, to avoid transfer state again
         if (code != TSDB_CODE_SUCCESS) { // todo handle this
           return 0;
         }
@@ -611,3 +614,13 @@ int32_t streamTaskReloadState(SStreamTask* pTask) {
     return TSDB_CODE_SUCCESS;
   }
 }
+
+int32_t streamAlignTransferState(SStreamTask* pTask) {
+  int32_t numOfUpstream = taosArrayGetSize(pTask->pUpstreamEpInfoList);
+  int32_t old = atomic_val_compare_exchange_32(&pTask->transferStateAlignCnt, 0, numOfUpstream);
+  if (old == 0) {
+    qDebug("s-task:%s set the transfer state aligncnt %d", pTask->id.idStr, numOfUpstream);
+  }
+
+  return atomic_sub_fetch_32(&pTask->transferStateAlignCnt, 1);
+}

source/libs/stream/src/streamMeta.c

@@ -264,8 +264,9 @@ SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId) {
   SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, &taskId, sizeof(int32_t));
   if (ppTask != NULL) {
     if (!streamTaskShouldStop(&(*ppTask)->status)) {
-      atomic_add_fetch_32(&(*ppTask)->refCnt, 1);
+      int32_t ref = atomic_add_fetch_32(&(*ppTask)->refCnt, 1);
       taosRUnLockLatch(&pMeta->lock);
+      qDebug("s-task:%s acquire task, ref:%d", (*ppTask)->id.idStr, ref);
       return *ppTask;
     }
   }
@@ -275,12 +276,24 @@ SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId) {
 }
 
 void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask) {
-  int32_t left = atomic_sub_fetch_32(&pTask->refCnt, 1);
-  if (left < 0) {
-    qError("task ref is invalid, ref:%d, %s", left, pTask->id.idStr);
-  } else if (left == 0) {
+  int32_t ref = atomic_sub_fetch_32(&pTask->refCnt, 1);
+  if (ref > 0) {
+    qDebug("s-task:%s release task, ref:%d", pTask->id.idStr, ref);
+  } else if (ref == 0) {
     ASSERT(streamTaskShouldStop(&pTask->status));
     tFreeStreamTask(pTask);
+  } else if (ref < 0) {
+    qError("task ref is invalid, ref:%d, %s", ref, pTask->id.idStr);
+  }
+}
+
+static void doRemoveIdFromList(SStreamMeta* pMeta, int32_t num, int32_t taskId) {
+  for (int32_t i = 0; i < num; ++i) {
+    int32_t* pTaskId = taosArrayGet(pMeta->pTaskList, i);
+    if (*pTaskId == taskId) {
+      taosArrayRemove(pMeta->pTaskList, i);
+      break;
+    }
   }
 }
 
@@ -333,17 +346,17 @@ void streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId) {
 
     int32_t num = taosArrayGetSize(pMeta->pTaskList);
     qDebug("s-task:%s set the drop task flag, remain running s-task:%d", pTask->id.idStr, num - 1);
-    for (int32_t i = 0; i < num; ++i) {
-      int32_t* pTaskId = taosArrayGet(pMeta->pTaskList, i);
-      if (*pTaskId == taskId) {
-        taosArrayRemove(pMeta->pTaskList, i);
-        break;
-      }
+    doRemoveIdFromList(pMeta, num, pTask->id.taskId);
+
+    // remove the ref by timer
+    if (pTask->triggerParam != 0) {
+      taosTmrStop(pTask->schedTimer);
+      streamMetaReleaseTask(pMeta, pTask);
     }
 
     streamMetaReleaseTask(pMeta, pTask);
   } else {
-    qDebug("vgId:%d failed to find the task:0x%x, it may be dropped already", pMeta->vgId, taskId);
+    qDebug("vgId:%d failed to find the task:0x%x, it may have been dropped already", pMeta->vgId, taskId);
   }
 
   taosWUnLockLatch(&pMeta->lock);

source/libs/stream/src/streamRecover.c

@@ -43,6 +43,7 @@ const char* streamGetTaskStatusStr(int32_t status) {
     case TASK_STATUS__SCAN_HISTORY: return "scan-history";
     case TASK_STATUS__HALT: return "halt";
     case TASK_STATUS__PAUSE: return "paused";
+    case TASK_STATUS__DROPPING: return "dropping";
     default:return "";
   }
 }
@@ -205,7 +206,7 @@ int32_t streamProcessCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* pRs
           qDebug("s-task:%s all %d downstream tasks are ready, now enter into scan-history-data stage, status:%s", id,
                  numOfReqs, streamGetTaskStatusStr(pTask->status.taskStatus));
           streamTaskLaunchScanHistory(pTask);
-        } else {  // todo add assert, agg tasks?
+        } else {
           ASSERT(pTask->status.taskStatus == TASK_STATUS__NORMAL);
           qDebug("s-task:%s fixed downstream task is ready, now ready for data from wal, status:%s", id,
                  streamGetTaskStatusStr(pTask->status.taskStatus));
@@ -258,9 +259,15 @@ int32_t streamRestoreParam(SStreamTask* pTask) {
 }
 
 int32_t streamSetStatusNormal(SStreamTask* pTask) {
-  qDebug("s-task:%s set task status to be normal, prev:%s", pTask->id.idStr, streamGetTaskStatusStr(pTask->status.taskStatus));
+  int32_t status = atomic_load_8(&pTask->status.taskStatus);
+  if (status == TASK_STATUS__DROPPING) {
+    qError("s-task:%s cannot be set normal, since in dropping state", pTask->id.idStr);
+    return -1;
+  } else {
+    qDebug("s-task:%s set task status to be normal, prev:%s", pTask->id.idStr, streamGetTaskStatusStr(status));
     atomic_store_8(&pTask->status.taskStatus, TASK_STATUS__NORMAL);
     return 0;
+  }
 }
 
 // source
@@ -344,7 +351,8 @@ static int32_t doDispatchTransferMsg(SStreamTask* pTask, const SStreamTransferRe
   msg.info.noResp = 1;
 
   tmsgSendReq(pEpSet, &msg);
-  qDebug("s-task:%s dispatch transfer state msg to taskId:0x%x (vgId:%d)", pTask->id.idStr, pReq->taskId, vgId);
+  qDebug("s-task:%s level:%d, status:%s dispatch transfer state msg to taskId:0x%x (vgId:%d)", pTask->id.idStr,
+         pTask->info.taskLevel, streamGetTaskStatusStr(pTask->status.taskStatus), pReq->taskId, vgId);
 
   return 0;
 }
@@ -354,9 +362,6 @@ int32_t streamDispatchTransferStateMsg(SStreamTask* pTask) {
 
   // serialize
   if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH) {
-    qDebug("s-task:%s send transfer state msg to downstream (fix-dispatch) to taskId:0x%x, status:%s", pTask->id.idStr,
-           pTask->fixedEpDispatcher.taskId, streamGetTaskStatusStr(pTask->status.taskStatus));
-
     req.taskId = pTask->fixedEpDispatcher.taskId;
     doDispatchTransferMsg(pTask, &req, pTask->fixedEpDispatcher.nodeId, &pTask->fixedEpDispatcher.epSet);
   } else if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
@@ -451,6 +456,7 @@ static void tryLaunchHistoryTask(void* param, void* tmrId) {
       const char* pStatus = streamGetTaskStatusStr((*ppTask)->status.taskStatus);
       qDebug("s-task:%s status:%s quit timer task", (*ppTask)->id.idStr, pStatus);
 
+      taosMemoryFree(pInfo);
       (*ppTask)->status.timerActive = 0;
       taosWUnLockLatch(&pMeta->lock);
       return;
@@ -511,6 +517,7 @@ int32_t streamCheckHistoryTaskDownstream(SStreamTask* pTask) {
       pTask->launchTaskTimer = taosTmrStart(tryLaunchHistoryTask,  100, pInfo, streamEnv.timer);
       if (pTask->launchTaskTimer == NULL) {
         // todo failed to create timer
+        taosMemoryFree(pInfo);
       } else {
         pTask->status.timerActive = 1;  // timer is active
         qDebug("s-task:%s set timer active flag", pTask->id.idStr);
@@ -553,8 +560,10 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask) {
   streamSetStatusNormal(pTask);
   atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
 
-  // todo check rsp, commit data
+  taosWLockLatch(&pMeta->lock);
   streamMetaSaveTask(pMeta, pTask);
+  taosWUnLockLatch(&pMeta->lock);
+
   return 0;
 }
 

source/libs/stream/src/streamTask.c

@@ -205,13 +205,16 @@ int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
 
 void tFreeStreamTask(SStreamTask* pTask) {
   qDebug("free s-task:%s", pTask->id.idStr);
+
   int32_t status = atomic_load_8((int8_t*)&(pTask->status.taskStatus));
   if (pTask->inputQueue) {
     streamQueueClose(pTask->inputQueue);
   }
+
   if (pTask->outputQueue) {
     streamQueueClose(pTask->outputQueue);
   }
+
   if (pTask->exec.qmsg) {
     taosMemoryFree(pTask->exec.qmsg);
   }
@@ -230,9 +233,7 @@ void tFreeStreamTask(SStreamTask* pTask) {
     tDeleteSchemaWrapper(pTask->tbSink.pSchemaWrapper);
     taosMemoryFree(pTask->tbSink.pTSchema);
     tSimpleHashCleanup(pTask->tbSink.pTblInfo);
-  }
-
-  if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
+  } else if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
     taosArrayDestroy(pTask->shuffleDispatcher.dbInfo.pVgroupInfos);
     taosArrayDestroy(pTask->checkReqIds);
     pTask->checkReqIds = NULL;

tests/parallel_test/cases.task

@@ -207,6 +207,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/precisionUS.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/precisionNS.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show_tag_index.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/information_schema.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py -R

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

@@ -87,6 +87,14 @@ class TDTestCase:
         for key,value in sql_dict.items():
             create_sql += f' {key} {value}'
         return create_sql
+
+    def show_create_sysdb_sql(self):
+        sysdb_list = {'information_schema', 'performance_schema'}
+        for db in sysdb_list:
+          tdSql.query(f'show create database {db}')
+          tdSql.checkEqual(f'{db}',tdSql.queryResult[0][0])
+          tdSql.checkEqual(f'CREATE DATABASE `{db}`',tdSql.queryResult[0][1])
+
     def show_create_sql(self):
         create_db_sql = self.set_create_database_sql(self.db_param)
         print(create_db_sql)
@@ -191,6 +199,7 @@ class TDTestCase:
         self.ins_check()
         self.perf_check()
         self.show_create_sql()
+        self.show_create_sysdb_sql()
 
     def stop(self):
         tdSql.close()

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

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import re
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+from util.sqlset import *
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+
+    def check_tags(self):
+        tdSql.checkRows(2)
+        tdSql.checkCols(6)
+        tdSql.checkData(0, 0, 'ctb1')
+        tdSql.checkData(0, 1, 'db')
+        tdSql.checkData(0, 2, 'stb')
+        tdSql.checkData(0, 3, 't0')
+        tdSql.checkData(0, 4, 'INT')
+        tdSql.checkData(0, 5, 1)
+        tdSql.checkData(1, 0, 'ctb1')
+        tdSql.checkData(1, 1, 'db')
+        tdSql.checkData(1, 2, 'stb')
+        tdSql.checkData(1, 3, 't1')
+        tdSql.checkData(1, 4, 'INT')
+        tdSql.checkData(1, 5, 1)
+
+    def check_table_tags(self, is_super_table):
+
+        if is_super_table == False:
+            tdSql.checkRows(1)
+            tdSql.checkCols(3)
+            tdSql.checkData(0, 0, 'ctb1')
+            tdSql.checkData(0, 1, 1)
+            tdSql.checkData(0, 2, 1)
+        else:
+            tdSql.checkRows(2)
+            tdSql.checkCols(3)
+            tdSql.checkData(0, 0, 'ctb1')
+            tdSql.checkData(1, 0, 'ctb2')
+            tdSql.checkData(0, 1, 1)
+            tdSql.checkData(1, 1, 2)
+            tdSql.checkData(0, 2, 1)
+            tdSql.checkData(1, 2, 2)
+
+    def check_indexes(self):
+        tdSql.checkRows(1)
+        tdSql.checkCols(7)
+        tdSql.checkData(0, 0, 'idx1')
+        tdSql.checkData(0, 1, 'db')
+        tdSql.checkData(0, 2, 'stb')
+        tdSql.checkData(0, 3, -1)
+        tdSql.checkData(0, 5, 't1')
+        tdSql.checkData(0, 6, 'tag_index')
+
+    def run(self):
+        tdSql.execute(f'create database db')
+        tdSql.execute(f'use db')
+        tdSql.execute(f'create table stb (ts timestamp, c0 int) tags (t0 int, t1 int)')
+        tdSql.execute(f'create table ctb1 using stb tags (1, 1)')
+        tdSql.execute(f'create table ctb2 using stb tags (2, 2)')
+        tdSql.execute(f'insert into ctb1 values (now, 1)')
+        tdSql.execute(f'insert into ctb2 values (now, 2)')
+
+        # show tags
+        tdSql.query(f'show tags from stb')
+        tdSql.checkRows(0)
+        tdSql.query(f'show tags from stb')
+        tdSql.checkRows(0);
+        tdSql.query(f'show tags from `stb`')
+        tdSql.checkRows(0);
+        tdSql.query(f'show tags from stb from db')
+        tdSql.checkRows(0);
+        tdSql.query(f'show tags from `stb` from `db`')
+        tdSql.checkRows(0);
+        tdSql.query(f'show tags from db.stb')
+        tdSql.checkRows(0);
+        tdSql.query(f'show tags from `db`.`stb`')
+        tdSql.checkRows(0);
+        tdSql.query(f'show tags from ctb1')
+        self.check_tags();
+        tdSql.query(f'show tags from `ctb1`')
+        self.check_tags();
+        tdSql.query(f'show tags from ctb1 from db')
+        self.check_tags();
+        tdSql.query(f'show tags from `ctb1` from `db`')
+        self.check_tags();
+        tdSql.query(f'show tags from db.ctb1')
+        self.check_tags();
+        tdSql.query(f'show tags from `db`.`ctb1`')
+        self.check_tags();
+
+        tdSql.error(f'show tags from db.stb from db')
+        tdSql.error(f'show tags from `db`.`stb` from db')
+        tdSql.error(f'show tags from db.ctb1 from db')
+        tdSql.error(f'show tags from `db`.`ctb1` from db')
+
+        # show table tags
+        tdSql.query(f'show table tags from stb')
+        self.check_table_tags(True);
+        tdSql.query(f'show table tags from `stb`')
+        self.check_table_tags(True);
+        tdSql.query(f'show table tags from stb from db')
+        self.check_table_tags(True);
+        tdSql.query(f'show table tags from `stb` from `db`')
+        self.check_table_tags(True);
+        tdSql.query(f'show table tags from db.stb')
+        self.check_table_tags(True);
+        tdSql.query(f'show table tags from `db`.`stb`')
+        self.check_table_tags(True);
+
+        tdSql.query(f'show table tags from ctb1')
+        self.check_table_tags(False);
+        tdSql.query(f'show table tags from `ctb1`')
+        self.check_table_tags(False);
+        tdSql.query(f'show table tags from ctb1 from db')
+        self.check_table_tags(False);
+        tdSql.query(f'show table tags from `ctb1` from `db`')
+        self.check_table_tags(False);
+        tdSql.query(f'show table tags from db.ctb1')
+        self.check_table_tags(False);
+        tdSql.query(f'show table tags from `db`.`ctb1`')
+        self.check_table_tags(False);
+
+        tdSql.error(f'show table tags from db.stb from db')
+        tdSql.error(f'show table tags from `db`.`stb` from db')
+        tdSql.error(f'show table tags from db.ctb1 from db')
+        tdSql.error(f'show table tags from `db`.`ctb1` from db')
+
+        # show indexes
+        tdSql.execute(f'create index idx1 on stb (t1)')
+
+        tdSql.query(f'show indexes from stb')
+        self.check_indexes();
+        tdSql.query(f'show indexes from `stb`')
+        self.check_indexes();
+        tdSql.query(f'show indexes from stb from db')
+        self.check_indexes();
+        tdSql.query(f'show indexes from `stb` from `db`')
+        self.check_indexes();
+        tdSql.query(f'show indexes from db.stb')
+        self.check_indexes();
+        tdSql.query(f'show indexes from `db`.`stb`')
+        self.check_indexes();
+
+        tdSql.query(f'show indexes from ctb1')
+        tdSql.checkRows(0)
+        tdSql.query(f'show indexes from `ctb1`')
+        tdSql.checkRows(0)
+        tdSql.query(f'show indexes from ctb1 from db')
+        tdSql.checkRows(0)
+        tdSql.query(f'show indexes from `ctb1` from `db`')
+        tdSql.checkRows(0)
+        tdSql.query(f'show indexes from db.ctb1')
+        tdSql.checkRows(0)
+        tdSql.query(f'show indexes from `db`.`ctb1`')
+        tdSql.checkRows(0)
+
+        tdSql.error(f'show indexes from db.stb from db')
+        tdSql.error(f'show indexes from `db`.`stb` from db')
+        tdSql.error(f'show indexes from db.ctb1 from db')
+        tdSql.error(f'show indexes from `db`.`ctb1` from db')
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())