xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

opt tbname in

This commit is contained in:
yihaoDeng 2022-09-30 10:19:56 +08:00
parent 0d3da10954 e6ac8ed3a9
commit f41b494ac6
86 changed files with 3862 additions and 3285 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/cmake.version
View File

@ -2,7 +2,7 @@
IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER})
ELSE ()
SET(TD_VER_NUMBER "3.0.1.3")
SET(TD_VER_NUMBER "3.0.1.4")
ENDIF ()
IF (DEFINED VERCOMPATIBLE)

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

@ -109,7 +109,7 @@ TDengine's JDBC URL specification format is:
For establishing connections, native connections differ slightly from REST connections.
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection">
```java

2
docs/en/14-reference/03-connector/05-go.mdx
View File

@ -113,7 +113,7 @@ username:password@protocol(address)/dbname?param=value
```
### Connecting via connector
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection">
_taosSql_ implements Go's `database/sql/driver` interface via cgo. You can use the [`database/sql`](https://golang.org/pkg/database/sql/) interface by simply introducing the driver.

22
docs/en/14-reference/03-connector/06-rust.mdx
View File

@ -55,16 +55,6 @@ taos = "*"
</TabItem>
<TabItem value="native" label="native connection only">
In `cargo.toml`, add [taos][taos] and enable the native feature:
```toml
[dependencies]
taos = { version = "*", default-features = false, features = ["native"] }
```
</TabItem>
<TabItem value="rest" label="Websocket only">
In `cargo.toml`, add [taos][taos] and enable the ws feature:
@ -75,6 +65,18 @@ taos = { version = "*", default-features = false, features = ["ws"] }
```
</TabItem>
<TabItem value="native" label="native connection only">
In `cargo.toml`, add [taos][taos] and enable the native feature:
```toml
[dependencies]
taos = { version = "*", default-features = false, features = ["native"] }
```
</TabItem>
</Tabs>
## Establishing a connection

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

@ -80,7 +80,7 @@ pip3 install git+https://github.com/taosdata/taos-connector-python.git
### Verify
<Tabs groupId="connect" default="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection">
For native connection, you need to verify that both the client driver and the Python connector itself are installed correctly. The client driver and Python connector have been installed properly if you can successfully import the `taos` module. In the Python Interactive Shell, you can type.
@ -118,7 +118,7 @@ Requirement already satisfied: taospy in c:\users\username\appdata\local\program
Before establishing a connection with the connector, we recommend testing the connectivity of the local TDengine CLI to the TDengine cluster.
<Tabs>
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection">
Ensure that the TDengine instance is up and that the FQDN of the machines in the cluster (the FQDN defaults to hostname if you are starting a standalone version) can be resolved locally, by testing with the `ping` command.
@ -173,7 +173,7 @@ If the test is successful, it will output the server version information, e.g.
The following example code assumes that TDengine is installed locally and that the default configuration is used for both FQDN and serverPort.
<Tabs>
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection" groupId="connect">
```python
@ -219,7 +219,7 @@ All arguments to the `connect()` function are optional keyword arguments. The fo
### Basic Usage
<Tabs default="native" groupId="connect">
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection">
##### TaosConnection class
@ -289,7 +289,7 @@ For a more detailed description of the `sql()` method, please refer to [RestClie
### Used with pandas
<Tabs default="native" groupId="connect">
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection">
```python

4
docs/en/14-reference/03-connector/08-node.mdx
View File

@ -85,7 +85,7 @@ If using ARM64 Node.js on Windows 10 ARM, you must add "Visual C++ compilers and
### Install via npm
<Tabs defaultValue="install_native">
<Tabs defaultValue="install_rest">
<TabItem value="install_native" label="Install native connector">
```bash
@ -124,7 +124,7 @@ node nodejsChecker.js host=localhost
Please choose to use one of the connectors.
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="native connection">
Install and import the `@tdengine/client` package.

8
docs/en/14-reference/03-connector/09-csharp.mdx
View File

@ -97,7 +97,7 @@ dotnet add exmaple.csproj reference src/TDengine.csproj
## Establish a Connection
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="Native Connection">
@ -173,7 +173,7 @@ ws://localhost:6041/test
#### SQL Write
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="Native Connection">
@ -204,7 +204,7 @@ ws://localhost:6041/test
#### Parameter Binding
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="Native Connection">
@ -227,7 +227,7 @@ ws://localhost:6041/test
#### Synchronous Query
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="Native Connection">

1
docs/en/14-reference/_telegraf.mdx
View File

@ -22,5 +22,4 @@ An example is as follows.
username = "root"
password = "taosdata"
data_format = "influx"
influx_max_line_bytes = 250
```

4
docs/en/20-third-party/02-prometheus.md vendored
View File

@ -89,3 +89,7 @@ VALUE TIMESTAMP
```
:::note
- TDengine will automatically create unique IDs for sub-table names by the rule.
:::

20
docs/en/20-third-party/03-telegraf.md vendored
View File

@ -15,6 +15,7 @@ To write Telegraf data to TDengine requires the following preparations.
- The TDengine cluster is deployed and functioning properly
- taosAdapter is installed and running properly. Please refer to the [taosAdapter manual](/reference/taosadapter) for details.
- Telegraf has been installed. Please refer to the [official documentation](https://docs.influxdata.com/telegraf/v1.22/install/) for Telegraf installation.
- Telegraf collects the running status measurements of current system. You can enable [input plugins](https://docs.influxdata.com/telegraf/v1.22/plugins/) to insert [other formats](https://docs.influxdata.com/telegraf/v1.24/data_formats/input/) data to Telegraf then forward to TDengine.
## Configuration steps
<Telegraf />
@ -31,11 +32,12 @@ Use TDengine CLI to verify Telegraf correctly writing data to TDengine and read
```
taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status |
====================================================================================================================================================================================================================================================================================
telegraf | 2022-04-20 08:47:53.488 | 22 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready |
log | 2022-04-20 07:19:50.260 | 9 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready |
Query OK, 2 row(s) in set (0.002401s)
name |
=================================
information_schema |
performance_schema |
telegraf |
Query OK, 3 rows in database (0.010568s)
taos> use telegraf;
Database changed.
@ -65,3 +67,11 @@ taos> select * from telegraf.system limit 10;
|
Query OK, 3 row(s) in set (0.013269s)
```
:::note
- TDengine take influxdb format data and create unique ID for table names by the rule.
The user can configure `smlChildTableName` parameter to generate specified table names if he/she needs. And he/she also need to insert data with specified data format.
For example, Add `smlChildTableName=tname` in the taos.cfg file. Insert data `st,tname=cpu1,t1=4 c1=3 1626006833639000000` then the table name will be cpu1. If there are multiple lines has same tname but different tag_set, the first line's tag_set will be used to automatically creating table and ignore other lines. Please refer to [TDengine Schemaless](/reference/schemaless/#Schemaless-Line-Protocol)
:::

4
docs/en/20-third-party/05-collectd.md vendored
View File

@ -72,3 +72,7 @@ taos> select * from collectd.memory_value limit 10;
Query OK, 10 row(s) in set (0.010348s)
```
:::note
- TDengine will automatically create unique IDs for sub-table names by the rule.
:::

16
docs/en/20-third-party/07-icinga2.md vendored
View File

@ -36,11 +36,12 @@ After waiting about 10 seconds, use the TDengine CLI to query TDengine to verify
```
taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status |
====================================================================================================================================================================================================================================================================================
log | 2022-04-20 07:19:50.260 | 11 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready |
icinga2 | 2022-04-20 12:11:39.697 | 13 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready |
Query OK, 2 row(s) in set (0.001867s)
name |
=================================
information_schema |
performance_schema |
icinga2 |
Query OK, 3 row(s) in set (0.001867s)
taos> use icinga2;
Database changed.
@ -72,3 +73,8 @@ taos> show stables;
icinga.service.users.state | 2022-04-20 12:11:39.704 | 2 | 1 | 1 |
Query OK, 22 row(s) in set (0.002317s)
```
:::note
- TDengine will automatically create unique IDs for sub-table names by the rule.
:::

1
docs/en/25-application/01-telegraf.md
View File

@ -60,7 +60,6 @@ For the configuration method, add the following text to `/etc/telegraf/telegraf.
username = "<TDengine's username>"
password = "<TDengine's password>"
data_format = "influx"
influx_max_line_bytes = 250
```
Then restart telegraf:

4
docs/en/28-releases/01-tdengine.md
View File

@ -6,6 +6,10 @@ description: TDengine release history, Release Notes and download links.
import Release from "/components/ReleaseV3";
## 3.0.1.4
<Release type="tdengine" version="3.0.1.4" />
## 3.0.1.3
<Release type="tdengine" version="3.0.1.3" />

4
docs/en/28-releases/02-tools.md
View File

@ -6,6 +6,10 @@ description: taosTools release history, Release Notes, download links.
import Release from "/components/ReleaseV3";
## 2.2.4
<Release type="tools" version="2.2.4" />
## 2.2.3
<Release type="tools" version="2.2.3" />

2
docs/zh/08-connector/14-java.mdx
View File

@ -109,7 +109,7 @@ TDengine 的 JDBC URL 规范格式为：
对于建立连接，原生连接与 REST 连接有细微不同。
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
```java

2
docs/zh/08-connector/20-go.mdx
View File

@ -114,7 +114,7 @@ username:password@protocol(address)/dbname?param=value
```
### 使用连接器进行连接
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
_taosSql_ 通过 cgo 实现了 Go 的 `database/sql/driver` 接口。只需要引入驱动就可以使用 [`database/sql`](https://golang.org/pkg/database/sql/) 的接口。

21
docs/zh/08-connector/26-rust.mdx
View File

@ -55,16 +55,6 @@ taos = "*"
</TabItem>
<TabItem value="native" label="仅原生连接">
在 `Cargo.toml` 文件中添加 [taos][taos]，并启用 `native` 特性：
```toml
[dependencies]
taos = { version = "*", default-features = false, features = ["native"] }
```
</TabItem>
<TabItem value="rest" label="仅 Websocket">
在 `Cargo.toml` 文件中添加 [taos][taos]，并启用 `ws` 特性。
@ -74,6 +64,17 @@ taos = { version = "*", default-features = false, features = ["native"] }
taos = { version = "*", default-features = false, features = ["ws"] }
```
</TabItem>
<TabItem value="native" label="仅原生连接">
在 `Cargo.toml` 文件中添加 [taos][taos]，并启用 `native` 特性：
```toml
[dependencies]
taos = { version = "*", default-features = false, features = ["native"] }
```
</TabItem>
</Tabs>

10
docs/zh/08-connector/30-python.mdx
View File

@ -80,7 +80,7 @@ pip3 install git+https://github.com/taosdata/taos-connector-python.git
### 安装验证
<Tabs groupId="connect" default="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
对于原生连接，需要验证客户端驱动和 Python 连接器本身是否都正确安装。如果能成功导入 `taos` 模块，则说明已经正确安装了客户端驱动和 Python 连接器。可在 Python 交互式 Shell 中输入：
@ -118,7 +118,7 @@ Requirement already satisfied: taospy in c:\users\username\appdata\local\program
在用连接器建立连接之前，建议先测试本地 TDengine CLI 到 TDengine 集群的连通性。
<Tabs>
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
请确保 TDengine 集群已经启动， 且集群中机器的 FQDN （如果启动的是单机版，FQDN 默认为 hostname）在本机能够解析, 可用 `ping` 命令进行测试：
@ -173,7 +173,7 @@ curl -u root:taosdata http://<FQDN>:<PORT>/rest/sql -d "select server_version()"
以下示例代码假设 TDengine 安装在本机， 且 FQDN 和 serverPort 都使用了默认配置。
<Tabs>
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接" groupId="connect">
```python
@ -219,7 +219,7 @@ curl -u root:taosdata http://<FQDN>:<PORT>/rest/sql -d "select server_version()"
### 基本使用
<Tabs default="native" groupId="connect">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
##### TaosConnection 类的使用
@ -289,7 +289,7 @@ TaosCursor 类使用原生连接进行写入、查询操作。在客户端多线
### 与 pandas 一起使用
<Tabs default="native" groupId="connect">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
```python

4
docs/zh/08-connector/35-node.mdx
View File

@ -85,7 +85,7 @@ REST 连接器支持所有能运行 Node.js 的平台。
### 使用 npm 安装
<Tabs defaultValue="install_native">
<Tabs defaultValue="install_rest">
<TabItem value="install_native" label="安装原生连接器">
```bash
@ -124,7 +124,7 @@ node nodejsChecker.js host=localhost
请选择使用一种连接器。
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
安装并引用 `@tdengine/client` 包。

10
docs/zh/08-connector/40-csharp.mdx
View File

@ -35,7 +35,7 @@ import CSAsyncQuery from "../07-develop/04-query-data/_cs_async.mdx"
## 支持的功能特性
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
@ -96,7 +96,7 @@ dotnet add exmaple.csproj reference src/TDengine.csproj
## 建立连接
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
@ -171,7 +171,7 @@ namespace TDengineExample
#### SQL 写入
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
@ -203,7 +203,7 @@ namespace TDengineExample
#### 参数绑定
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
@ -227,7 +227,7 @@ namespace TDengineExample
#### 同步查询
<Tabs defaultValue="native">
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">

1
docs/zh/14-reference/_telegraf.mdx
View File

@ -22,6 +22,5 @@
username = "root"
password = "taosdata"
data_format = "influx"
influx_max_line_bytes = 250
```

4
docs/zh/20-third-party/02-prometheus.md vendored
View File

@ -88,3 +88,7 @@ VALUE TIMESTAMP
```
:::note
- TDengine 默认生成的子表名是根据规则生成的唯一 ID 值。
:::

20
docs/zh/20-third-party/03-telegraf.md vendored
View File

@ -16,6 +16,7 @@ Telegraf 是一款十分流行的指标采集开源软件。在数据采集和
- TDengine 集群已经部署并正常运行
- taosAdapter 已经安装并正常运行。具体细节请参考 [taosAdapter 的使用手册](/reference/taosadapter)
- Telegraf 已经安装。安装 Telegraf 请参考[官方文档](https://docs.influxdata.com/telegraf/v1.22/install/)
- Telegraf 默认采集系统运行状态数据。通过使能[输入插件](https://docs.influxdata.com/telegraf/v1.22/plugins/)方式可以输出[其他格式](https://docs.influxdata.com/telegraf/v1.24/data_formats/input/)的数据到 Telegraf 再写入到 TDengine中。
## 配置步骤
<Telegraf />
@ -32,11 +33,12 @@ sudo systemctl restart telegraf
```
taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status |
====================================================================================================================================================================================================================================================================================
telegraf | 2022-04-20 08:47:53.488 | 22 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready |
log | 2022-04-20 07:19:50.260 | 9 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready |
Query OK, 2 row(s) in set (0.002401s)
name |
=================================
information_schema |
performance_schema |
telegraf |
Query OK, 3 rows in database (0.010568s)
taos> use telegraf;
Database changed.
@ -66,3 +68,11 @@ taos> select * from telegraf.system limit 10;
|
Query OK, 3 row(s) in set (0.013269s)
```
:::note
- TDengine 接收 influxdb 格式数据默认生成的子表名是根据规则生成的唯一 ID 值。
用户如需指定生成的表名，可以通过在 taos.cfg 里配置 smlChildTableName 参数来指定。如果通过控制输入数据格式，即可利用 TDengine 这个功能指定生成的表名。
举例如下：配置 smlChildTableName=tname 插入数据为 st,tname=cpu1,t1=4 c1=3 1626006833639000000 则创建的表名为 cpu1。如果多行数据 tname 相同，但是后面的 tag_set 不同，则使用第一行自动建表时指定的 tag_set，其他的行会忽略）。[TDengine 无模式写入参考指南](/reference/schemaless/#无模式写入行协议)
:::

4
docs/zh/20-third-party/05-collectd.md vendored
View File

@ -72,3 +72,7 @@ taos> select * from collectd.memory_value limit 10;
Query OK, 10 row(s) in set (0.010348s)
```
:::note
- TDengine 默认生成的子表名是根据规则生成的唯一 ID 值。
:::

17
docs/zh/20-third-party/07-icinga2.md vendored
View File

@ -37,11 +37,12 @@ sudo systemctl restart icinga2
```
taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status |
====================================================================================================================================================================================================================================================================================
log | 2022-04-20 07:19:50.260 | 11 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready |
icinga2 | 2022-04-20 12:11:39.697 | 13 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready |
Query OK, 2 row(s) in set (0.001867s)
name |
=================================
information_schema |
performance_schema |
icinga2 |
Query OK, 3 row(s) in set (0.001867s)
taos> use icinga2;
Database changed.
@ -73,3 +74,9 @@ taos> show stables;
icinga.service.users.state | 2022-04-20 12:11:39.704 | 2 | 1 | 1 |
Query OK, 22 row(s) in set (0.002317s)
```
:::note
- TDengine 默认生成的子表名是根据规则生成的唯一 ID 值。
:::

1
docs/zh/25-application/01-telegraf.md
View File

@ -61,7 +61,6 @@ IT 运维监测数据通常都是对时间特性比较敏感的数据，例如
username = "<TDengine's username>"
password = "<TDengine's password>"
data_format = "influx"
influx_max_line_bytes = 250
```
然后重启 Telegraf：

4
docs/zh/28-releases/01-tdengine.md
View File

@ -6,6 +6,10 @@ description: TDengine 发布历史、Release Notes 及下载链接
import Release from "/components/ReleaseV3";
## 3.0.1.4
<Release type="tdengine" version="3.0.1.4" />
## 3.0.1.3
<Release type="tdengine" version="3.0.1.3" />

4
docs/zh/28-releases/02-tools.md
View File

@ -6,6 +6,10 @@ description: taosTools 的发布历史、Release Notes 和下载链接
import Release from "/components/ReleaseV3";
## 2.2.4
<Release type="tools" version="2.2.4" />
## 2.2.3
<Release type="tools" version="2.2.3" />

1
include/common/tcommon.h
View File

@ -177,6 +177,7 @@ typedef struct SSDataBlock {
enum {
FETCH_TYPE__DATA = 1,
FETCH_TYPE__META,
FETCH_TYPE__SEP,
FETCH_TYPE__NONE,
};

209
include/common/ttokendef.h
View File

@ -227,110 +227,111 @@
#define TK_WEND 209
#define TK_WDURATION 210
#define TK_IROWTS 211
#define TK_CAST 212
#define TK_NOW 213
#define TK_TODAY 214
#define TK_TIMEZONE 215
#define TK_CLIENT_VERSION 216
#define TK_SERVER_VERSION 217
#define TK_SERVER_STATUS 218
#define TK_CURRENT_USER 219
#define TK_COUNT 220
#define TK_LAST_ROW 221
#define TK_CASE 222
#define TK_END 223
#define TK_WHEN 224
#define TK_THEN 225
#define TK_ELSE 226
#define TK_BETWEEN 227
#define TK_IS 228
#define TK_NK_LT 229
#define TK_NK_GT 230
#define TK_NK_LE 231
#define TK_NK_GE 232
#define TK_NK_NE 233
#define TK_MATCH 234
#define TK_NMATCH 235
#define TK_CONTAINS 236
#define TK_IN 237
#define TK_JOIN 238
#define TK_INNER 239
#define TK_SELECT 240
#define TK_DISTINCT 241
#define TK_WHERE 242
#define TK_PARTITION 243
#define TK_BY 244
#define TK_SESSION 245
#define TK_STATE_WINDOW 246
#define TK_SLIDING 247
#define TK_FILL 248
#define TK_VALUE 249
#define TK_NONE 250
#define TK_PREV 251
#define TK_LINEAR 252
#define TK_NEXT 253
#define TK_HAVING 254
#define TK_RANGE 255
#define TK_EVERY 256
#define TK_ORDER 257
#define TK_SLIMIT 258
#define TK_SOFFSET 259
#define TK_LIMIT 260
#define TK_OFFSET 261
#define TK_ASC 262
#define TK_NULLS 263
#define TK_ABORT 264
#define TK_AFTER 265
#define TK_ATTACH 266
#define TK_BEFORE 267
#define TK_BEGIN 268
#define TK_BITAND 269
#define TK_BITNOT 270
#define TK_BITOR 271
#define TK_BLOCKS 272
#define TK_CHANGE 273
#define TK_COMMA 274
#define TK_COMPACT 275
#define TK_CONCAT 276
#define TK_CONFLICT 277
#define TK_COPY 278
#define TK_DEFERRED 279
#define TK_DELIMITERS 280
#define TK_DETACH 281
#define TK_DIVIDE 282
#define TK_DOT 283
#define TK_EACH 284
#define TK_FAIL 285
#define TK_FILE 286
#define TK_FOR 287
#define TK_GLOB 288
#define TK_ID 289
#define TK_IMMEDIATE 290
#define TK_IMPORT 291
#define TK_INITIALLY 292
#define TK_INSTEAD 293
#define TK_ISNULL 294
#define TK_KEY 295
#define TK_NK_BITNOT 296
#define TK_NK_SEMI 297
#define TK_NOTNULL 298
#define TK_OF 299
#define TK_PLUS 300
#define TK_PRIVILEGE 301
#define TK_RAISE 302
#define TK_REPLACE 303
#define TK_RESTRICT 304
#define TK_ROW 305
#define TK_SEMI 306
#define TK_STAR 307
#define TK_STATEMENT 308
#define TK_STRING 309
#define TK_TIMES 310
#define TK_UPDATE 311
#define TK_VALUES 312
#define TK_VARIABLE 313
#define TK_VIEW 314
#define TK_WAL 315
#define TK_QTAGS 212
#define TK_CAST 213
#define TK_NOW 214
#define TK_TODAY 215
#define TK_TIMEZONE 216
#define TK_CLIENT_VERSION 217
#define TK_SERVER_VERSION 218
#define TK_SERVER_STATUS 219
#define TK_CURRENT_USER 220
#define TK_COUNT 221
#define TK_LAST_ROW 222
#define TK_CASE 223
#define TK_END 224
#define TK_WHEN 225
#define TK_THEN 226
#define TK_ELSE 227
#define TK_BETWEEN 228
#define TK_IS 229
#define TK_NK_LT 230
#define TK_NK_GT 231
#define TK_NK_LE 232
#define TK_NK_GE 233
#define TK_NK_NE 234
#define TK_MATCH 235
#define TK_NMATCH 236
#define TK_CONTAINS 237
#define TK_IN 238
#define TK_JOIN 239
#define TK_INNER 240
#define TK_SELECT 241
#define TK_DISTINCT 242
#define TK_WHERE 243
#define TK_PARTITION 244
#define TK_BY 245
#define TK_SESSION 246
#define TK_STATE_WINDOW 247
#define TK_SLIDING 248
#define TK_FILL 249
#define TK_VALUE 250
#define TK_NONE 251
#define TK_PREV 252
#define TK_LINEAR 253
#define TK_NEXT 254
#define TK_HAVING 255
#define TK_RANGE 256
#define TK_EVERY 257
#define TK_ORDER 258
#define TK_SLIMIT 259
#define TK_SOFFSET 260
#define TK_LIMIT 261
#define TK_OFFSET 262
#define TK_ASC 263
#define TK_NULLS 264
#define TK_ABORT 265
#define TK_AFTER 266
#define TK_ATTACH 267
#define TK_BEFORE 268
#define TK_BEGIN 269
#define TK_BITAND 270
#define TK_BITNOT 271
#define TK_BITOR 272
#define TK_BLOCKS 273
#define TK_CHANGE 274
#define TK_COMMA 275
#define TK_COMPACT 276
#define TK_CONCAT 277
#define TK_CONFLICT 278
#define TK_COPY 279
#define TK_DEFERRED 280
#define TK_DELIMITERS 281
#define TK_DETACH 282
#define TK_DIVIDE 283
#define TK_DOT 284
#define TK_EACH 285
#define TK_FAIL 286
#define TK_FILE 287
#define TK_FOR 288
#define TK_GLOB 289
#define TK_ID 290
#define TK_IMMEDIATE 291
#define TK_IMPORT 292
#define TK_INITIALLY 293
#define TK_INSTEAD 294
#define TK_ISNULL 295
#define TK_KEY 296
#define TK_NK_BITNOT 297
#define TK_NK_SEMI 298
#define TK_NOTNULL 299
#define TK_OF 300
#define TK_PLUS 301
#define TK_PRIVILEGE 302
#define TK_RAISE 303
#define TK_REPLACE 304
#define TK_RESTRICT 305
#define TK_ROW 306
#define TK_SEMI 307
#define TK_STAR 308
#define TK_STATEMENT 309
#define TK_STRING 310
#define TK_TIMES 311
#define TK_UPDATE 312
#define TK_VALUES 313
#define TK_VARIABLE 314
#define TK_VIEW 315
#define TK_WAL 316
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301

13
include/libs/executor/executor.h
View File

@ -29,13 +29,13 @@ typedef void* DataSinkHandle;
struct SRpcMsg;
struct SSubplan;
typedef int32_t (*localFetchFp)(void *, uint64_t, uint64_t, uint64_t, int64_t, int32_t, void**, SArray*);
typedef int32_t (*localFetchFp)(void*, uint64_t, uint64_t, uint64_t, int64_t, int32_t, void**, SArray*);
typedef struct {
void *handle;
void* handle;
bool localExec;
localFetchFp fp;
SArray *explainRes;
SArray* explainRes;
} SLocalFetch;
typedef struct {
@ -51,9 +51,9 @@ typedef struct {
bool initTqReader;
int32_t numOfVgroups;
void* sContext; // SSnapContext*
void* sContext; // SSnapContext*
void* pStateBackend;
void* pStateBackend;
} SReadHandle;
// in queue mode, data streams are seperated by msg
@ -136,6 +136,7 @@ int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* table
* @param handle
* @return
*/
int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bool* hasMore, SLocalFetch *pLocal);
int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pBlock, uint64_t* useconds);
@ -195,6 +196,8 @@ int32_t qStreamPrepareTsdbScan(qTaskInfo_t tinfo, uint64_t uid, int64_t ts);
int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType);
int32_t qStreamScanMemData(qTaskInfo_t tinfo, const SSubmitReq* pReq);
int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo);

1
include/libs/function/functionMgt.h
View File

@ -120,6 +120,7 @@ typedef enum EFunctionType {
FUNCTION_TYPE_WEND,
FUNCTION_TYPE_WDURATION,
FUNCTION_TYPE_IROWTS,
FUNCTION_TYPE_TAGS,
// internal function
FUNCTION_TYPE_SELECT_VALUE = 3750,

8
include/libs/nodes/nodes.h
View File

@ -27,9 +27,10 @@ extern "C" {
#define LIST_LENGTH(l) (NULL != (l) ? (l)->length : 0)
#define FOREACH(node, list) \
for (SListCell *cell = (NULL != (list) ? (list)->pHead : NULL), *pNext; \
(NULL != cell ? (node = cell->pNode, pNext = cell->pNext, true) : (node = NULL, pNext = NULL, false)); cell = pNext)
#define FOREACH(node, list) \
for (SListCell* cell = (NULL != (list) ? (list)->pHead : NULL), *pNext; \
(NULL != cell ? (node = cell->pNode, pNext = cell->pNext, true) : (node = NULL, pNext = NULL, false)); \
cell = pNext)
#define REPLACE_NODE(newNode) cell->pNode = (SNode*)(newNode)
@ -192,6 +193,7 @@ typedef enum ENodeType {
QUERY_NODE_SHOW_TABLE_DISTRIBUTED_STMT,
QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT,
QUERY_NODE_SHOW_SCORES_STMT,
QUERY_NODE_SHOW_TABLE_TAGS_STMT,
QUERY_NODE_KILL_CONNECTION_STMT,
QUERY_NODE_KILL_QUERY_STMT,
QUERY_NODE_KILL_TRANSACTION_STMT,

2
include/libs/sync/sync.h
View File

@ -132,7 +132,7 @@ typedef struct SSyncFSM {
void (*FpRollBackCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
void (*FpRestoreFinishCb)(struct SSyncFSM* pFsm);
void (*FpReConfigCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta);
void (*FpReConfigCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta *cbMeta);
void (*FpLeaderTransferCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
void (*FpBecomeLeaderCb)(struct SSyncFSM* pFsm);

1
include/util/taoserror.h
View File

@ -565,6 +565,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_GET_META_ERROR TAOS_DEF_ERROR_CODE(0, 0x2662)
#define TSDB_CODE_PAR_NOT_UNIQUE_TABLE_ALIAS TAOS_DEF_ERROR_CODE(0, 0x2663)
#define TSDB_CODE_PAR_NOT_SUPPORT_JOIN TAOS_DEF_ERROR_CODE(0, 0x2664)
#define TSDB_CODE_PAR_INVALID_TAGS_PC TAOS_DEF_ERROR_CODE(0, 0x2665)
#define TSDB_CODE_PAR_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x26FF)
//planner

4
packaging/release.sh
View File

@ -219,12 +219,12 @@ fi
if [[ "$cpuType" == "x64" ]] || [[ "$cpuType" == "aarch64" ]] || [[ "$cpuType" == "aarch32" ]] || [[ "$cpuType" == "arm64" ]] || [[ "$cpuType" == "arm32" ]] || [[ "$cpuType" == "mips64" ]]; then
if [ "$verMode" != "cluster" ]; then
# community-version compile
cmake ../ -DCPUTYPE=${cpuType} -DOSTYPE=${osType} -DSOMODE=${soMode} -DDBNAME=${dbName} -DVERTYPE=${verType} -DVERDATE="${build_time}" -DGITINFO=${gitinfo} -DGITINFOI=${gitinfoOfInternal} -DVERNUMBER=${verNumber} -DVERCOMPATIBLE=${verNumberComp} -DPAGMODE=${pagMode} -DBUILD_HTTP=${BUILD_HTTP} -DBUILD_TOOLS=${BUILD_TOOLS} ${allocator_macro}
cmake ../ -DCPUTYPE=${cpuType} -DWEBSOCKET=true -DOSTYPE=${osType} -DSOMODE=${soMode} -DDBNAME=${dbName} -DVERTYPE=${verType} -DVERDATE="${build_time}" -DGITINFO=${gitinfo} -DGITINFOI=${gitinfoOfInternal} -DVERNUMBER=${verNumber} -DVERCOMPATIBLE=${verNumberComp} -DPAGMODE=${pagMode} -DBUILD_HTTP=${BUILD_HTTP} -DBUILD_TOOLS=${BUILD_TOOLS} ${allocator_macro}
else
if [[ "$dbName" != "taos" ]]; then
replace_enterprise_$dbName
fi
cmake ../../ -DCPUTYPE=${cpuType} -DOSTYPE=${osType} -DSOMODE=${soMode} -DDBNAME=${dbName} -DVERTYPE=${verType} -DVERDATE="${build_time}" -DGITINFO=${gitinfo} -DGITINFOI=${gitinfoOfInternal} -DVERNUMBER=${verNumber} -DVERCOMPATIBLE=${verNumberComp} -DBUILD_HTTP=${BUILD_HTTP} -DBUILD_TOOLS=${BUILD_TOOLS} ${allocator_macro}
cmake ../../ -DCPUTYPE=${cpuType} -DWEBSOCKET=true -DOSTYPE=${osType} -DSOMODE=${soMode} -DDBNAME=${dbName} -DVERTYPE=${verType} -DVERDATE="${build_time}" -DGITINFO=${gitinfo} -DGITINFOI=${gitinfoOfInternal} -DVERNUMBER=${verNumber} -DVERCOMPATIBLE=${verNumberComp} -DBUILD_HTTP=${BUILD_HTTP} -DBUILD_TOOLS=${BUILD_TOOLS} ${allocator_macro}
fi
else
echo "input cpuType=${cpuType} error!!!"

3
source/client/src/clientImpl.c
View File

@ -874,8 +874,6 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
STscObj* pTscObj = pRequest->pTscObj;
pRequest->code = code;
pRequest->metric.resultReady = taosGetTimestampUs();
if (pResult) {
destroyQueryExecRes(&pRequest->body.resInfo.execRes);
memcpy(&pRequest->body.resInfo.execRes, pResult, sizeof(*pResult));
@ -1061,7 +1059,6 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
}
pRequest->metric.planEnd = taosGetTimestampUs();
if (TSDB_CODE_SUCCESS == code && !pRequest->validateOnly) {
SArray* pNodeList = NULL;
buildAsyncExecNodeList(pRequest, &pNodeList, pMnodeList, pResultMeta);

2
source/client/src/clientMain.c
View File

@ -817,7 +817,6 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
pRequest->metric.syntaxEnd = taosGetTimestampUs();
if (!updateMetaForce) {
STscObj *pTscObj = pRequest->pTscObj;
SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
if (NULL == pRequest->pQuery->pRoot) {
atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);
@ -864,6 +863,7 @@ static void fetchCallback(void *pResult, void *param, int32_t code) {
SRequestObj *pRequest = (SRequestObj *)param;
SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
pRequest->metric.resultReady = taosGetTimestampUs();
tscDebug("0x%" PRIx64 " enter scheduler fetch cb, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code,
tstrerror(code), pRequest->requestId);

24
source/client/src/clientTmq.c
View File

@ -515,7 +515,7 @@ int32_t tmqCommitMsgImpl(tmq_t* tmq, const TAOS_RES* msg, int8_t async, tmq_comm
SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)msg;
topic = pMetaRspObj->topic;
vgId = pMetaRspObj->vgId;
} else if(TD_RES_TMQ_METADATA(msg)) {
} else if (TD_RES_TMQ_METADATA(msg)) {
SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)msg;
topic = pRspObj->topic;
vgId = pRspObj->vgId;
@ -715,7 +715,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
int32_t epoch = tmq->epoch;
SMqHbReq* pReq = taosMemoryMalloc(sizeof(SMqHbReq));
if (pReq == NULL) goto OVER;
pReq->consumerId = consumerId;
pReq->consumerId = htobe64(consumerId);
pReq->epoch = epoch;
SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
@ -1603,6 +1603,7 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
return NULL;
} else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_RSP) {
SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)rspWrapper;
tscDebug("consumer %ld actual process poll rsp", tmq->consumerId);
/*atomic_sub_fetch_32(&tmq->readyRequest, 1);*/
int32_t consumerEpoch = atomic_load_32(&tmq->epoch);
if (pollRspWrapper->dataRsp.head.epoch == consumerEpoch) {
@ -1661,9 +1662,9 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
// build rsp
void* pRsp = NULL;
if(pollRspWrapper->taosxRsp.createTableNum == 0){
if (pollRspWrapper->taosxRsp.createTableNum == 0) {
pRsp = tmqBuildRspFromWrapper(pollRspWrapper);
}else{
} else {
pRsp = tmqBuildTaosxRspFromWrapper(pollRspWrapper);
}
taosFreeQitem(pollRspWrapper);
@ -1718,7 +1719,10 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
while (1) {
tmqHandleAllDelayedTask(tmq);
if (tmqPollImpl(tmq, timeout) < 0) return NULL;
if (tmqPollImpl(tmq, timeout) < 0) {
tscDebug("return since poll err");
/*return NULL;*/
}
rspObj = tmqHandleAllRsp(tmq, timeout, false);
if (rspObj) {
@ -1850,12 +1854,12 @@ const char* tmq_get_table_name(TAOS_RES* res) {
return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter);
} else if (TD_RES_TMQ_METADATA(res)) {
SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res;
if (!pRspObj->rsp.withTbName || pRspObj->rsp.blockTbName == NULL || pRspObj->resIter < 0 ||
pRspObj->resIter >= pRspObj->rsp.blockNum) {
return NULL;
}
return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter);
if (!pRspObj->rsp.withTbName || pRspObj->rsp.blockTbName == NULL || pRspObj->resIter < 0 ||
pRspObj->resIter >= pRspObj->rsp.blockNum) {
return NULL;
}
return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter);
}
return NULL;
}

3
source/dnode/mgmt/node_mgmt/src/dmProc.c
View File

@ -74,11 +74,14 @@ static SProcQueue *dmInitProcQueue(SProc *proc, char *ptr, int32_t size) {
}
tstrncpy(queue->name, proc->name, sizeof(queue->name));
taosThreadMutexLock(&queue->mutex);
queue->head = 0;
queue->tail = 0;
queue->total = bufSize;
queue->avail = bufSize;
queue->items = 0;
taosThreadMutexUnlock(&queue->mutex);
}
return queue;

2
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -301,7 +301,7 @@ int32_t dmInitServer(SDnode *pDnode) {
SDnodeTrans *pTrans = &pDnode->trans;
SRpcInit rpcInit = {0};
strncpy(rpcInit.localFqdn, tsLocalFqdn, TSDB_FQDN_LEN);
tstrncpy(rpcInit.localFqdn, tsLocalFqdn, TSDB_FQDN_LEN);
rpcInit.localPort = tsServerPort;
rpcInit.label = "DND-S";
rpcInit.numOfThreads = tsNumOfRpcThreads;

1
source/dnode/mnode/impl/src/mndAcct.c
View File

@ -83,6 +83,7 @@ static int32_t mndCreateDefaultAcct(SMnode *pMnode) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, NULL, "create-acct");
if (pTrans == NULL) {
sdbFreeRaw(pRaw);
mError("acct:%s, failed to create since %s", acctObj.acct, terrstr());
return -1;
}

2
source/dnode/mnode/impl/src/mndCluster.c
View File

@ -315,7 +315,7 @@ static int32_t mndProcessUptimeTimer(SRpcMsg *pReq) {
return 0;
}
mInfo("update cluster uptime to %" PRId64, clusterObj.upTime);
mInfo("update cluster uptime to %d", clusterObj.upTime);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "update-uptime");
if (pTrans == NULL) return -1;

1
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -272,6 +272,7 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
if (pConsumer == NULL) {
mError("consumer %ld not exist", consumerId);
terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
return -1;
}

17
source/dnode/mnode/impl/src/mndSync.c
View File

@ -68,7 +68,7 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM
if (pMgmt->errCode != 0) {
mError("trans:%d, failed to propose since %s, post sem", transId, tstrerror(pMgmt->errCode));
} else {
mInfo("trans:%d, is proposed and post sem", transId, tstrerror(pMgmt->errCode));
mInfo("trans:%d, is proposed and post sem", transId);
}
pMgmt->transId = 0;
taosWUnLockLatch(&pMgmt->lock);
@ -113,13 +113,13 @@ void mndRestoreFinish(struct SSyncFSM *pFsm) {
}
}
void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta cbMeta) {
void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta *cbMeta) {
SMnode *pMnode = pFsm->data;
SSyncMgmt *pMgmt = &pMnode->syncMgmt;
pMgmt->errCode = cbMeta.code;
pMgmt->errCode = cbMeta->code;
mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64, pMgmt->transId,
cbMeta.code, cbMeta.index, cbMeta.term);
cbMeta->code, cbMeta->index, cbMeta->term);
taosWLockLatch(&pMgmt->lock);
if (pMgmt->transId == -1) {
@ -127,7 +127,7 @@ void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta cbM
mError("trans:-1, failed to propose sync reconfig since %s, post sem", tstrerror(pMgmt->errCode));
} else {
mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64 " post sem",
pMgmt->transId, cbMeta.code, cbMeta.index, cbMeta.term);
pMgmt->transId, cbMeta->code, cbMeta->index, cbMeta->term);
}
pMgmt->transId = 0;
tsem_post(&pMgmt->syncSem);
@ -271,6 +271,11 @@ void mndCleanupSync(SMnode *pMnode) {
int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
SSyncMgmt *pMgmt = &pMnode->syncMgmt;
SRpcMsg req = {.msgType = TDMT_MND_APPLY_MSG, .contLen = sdbGetRawTotalSize(pRaw)};
if (req.contLen <= 0) {
terrno = TSDB_CODE_APP_ERROR;
return -1;
}
req.pCont = rpcMallocCont(req.contLen);
if (req.pCont == NULL) return -1;
memcpy(req.pCont, pRaw, req.contLen);
@ -278,7 +283,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
pMgmt->errCode = 0;
taosWLockLatch(&pMgmt->lock);
if (pMgmt->transId != 0) {
mError("trans:%d, can't be proposed since trans:%s alrady waiting for confirm", transId, pMgmt->transId);
mError("trans:%d, can't be proposed since trans:%d alrady waiting for confirm", transId, pMgmt->transId);
taosWUnLockLatch(&pMgmt->lock);
terrno = TSDB_CODE_APP_NOT_READY;
return -1;

4
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -379,6 +379,8 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
topicObj.ast = strdup(pCreate->ast);
topicObj.astLen = strlen(pCreate->ast) + 1;
qDebugL("ast %s", topicObj.ast);
SNode *pAst = NULL;
if (nodesStringToNode(pCreate->ast, &pAst) != 0) {
taosMemoryFree(topicObj.ast);
@ -774,7 +776,7 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
SName n;
int32_t cols = 0;
char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE + 5] = {0};
strcpy(varDataVal(topicName), mndGetDbStr(pTopic->name));
/*tNameFromString(&n, pTopic->name, T_NAME_ACCT | T_NAME_DB);*/
/*tNameGetDbName(&n, varDataVal(topicName));*/

6
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -427,7 +427,7 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
terrno = 0;
_OVER:
if (terrno != 0) {
if (terrno != 0 && pTrans != NULL) {
mError("trans:%d, failed to parse from raw:%p since %s", pTrans->id, pRaw, terrstr());
mndTransDropData(pTrans);
taosMemoryFreeClear(pRow);
@ -629,6 +629,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict,
pTrans->pRpcArray == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("failed to create transaction since %s", terrstr());
mndTransDrop(pTrans);
return NULL;
}
@ -1428,6 +1429,7 @@ static bool mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans) {
SSdbRaw *pRaw = mndTransActionEncode(pTrans);
if (pRaw == NULL) {
mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
return false;
}
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
@ -1617,7 +1619,7 @@ static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false);
char lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
char detail[TSDB_TRANS_ERROR_LEN] = {0};
char detail[TSDB_TRANS_ERROR_LEN + 1] = {0};
int32_t len = snprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
SEpSet epset = pTrans->lastEpset;

1
source/dnode/mnode/impl/src/mndUser.c
View File

@ -83,6 +83,7 @@ static int32_t mndCreateDefaultUser(SMnode *pMnode, char *acct, char *user, char
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, NULL, "create-user");
if (pTrans == NULL) {
sdbFreeRaw(pRaw);
mError("user:%s, failed to create since %s", userObj.user, terrstr());
return -1;
}

8
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -1250,7 +1250,7 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d avail:%" PRId64 " used:%" PRId64,
pVgroup->dbName, pVgroup->vgId, vgMem, pNew1->id, pNew1->memAvail, pNew1->memUsed);
terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
return -1;
goto _OVER;
} else {
pNew1->memUsed += vgMem;
}
@ -1272,7 +1272,7 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d avail:%" PRId64 " used:%" PRId64,
pVgroup->dbName, pVgroup->vgId, vgMem, pNew2->id, pNew2->memAvail, pNew2->memUsed);
terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
return -1;
goto _OVER;
} else {
pNew2->memUsed += vgMem;
}
@ -1293,7 +1293,7 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
mError("db:%s, vgId:%d, no enough memory:%" PRId64 " in dnode:%d avail:%" PRId64 " used:%" PRId64,
pVgroup->dbName, pVgroup->vgId, vgMem, pNew3->id, pNew3->memAvail, pNew3->memUsed);
terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
return -1;
goto _OVER;
} else {
pNew3->memUsed += vgMem;
}
@ -1627,7 +1627,7 @@ static int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj
} else if (newVg1.replica == 3) {
SVnodeGid del1 = {0};
if (mndRemoveVnodeFromVgroup(pMnode, &newVg1, pArray, &del1) != 0) goto _OVER;
if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del1, true) != 0) return -1;
if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del1, true) != 0) goto _OVER;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVg1, TDMT_VND_ALTER_REPLICA) != 0) goto _OVER;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVg1, &del1, true) != 0) goto _OVER;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg1) != 0) goto _OVER;

3
source/dnode/vnode/inc/vnode.h
View File

@ -94,6 +94,7 @@ typedef struct SMetaEntry SMetaEntry;
void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags);
void metaReaderClear(SMetaReader *pReader);
int32_t metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid);
int metaGetTableEntryByName(SMetaReader *pReader, const char *name);
int32_t metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags);
int32_t metaGetTableTagsOpt(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags);
int32_t metaReadNext(SMetaReader *pReader);
@ -222,7 +223,7 @@ int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
int32_t tqSeekVer(STqReader *pReader, int64_t ver);
int32_t tqNextBlock(STqReader *pReader, SFetchRet *ret);
int32_t tqReaderSetDataMsg(STqReader *pReader, SSubmitReq *pMsg, int64_t ver);
int32_t tqReaderSetDataMsg(STqReader *pReader, const SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReader *pReader);
bool tqNextDataBlockFilterOut(STqReader *pReader, SHashObj *filterOutUids);
int32_t tqRetrieveDataBlock(SSDataBlock *pBlock, STqReader *pReader);

18
source/dnode/vnode/src/inc/tq.h
View File

@ -113,10 +113,20 @@ typedef struct {
} STqHandle;
typedef struct {
SMqDataRsp dataRsp;
SMqRspHead rspHead;
char subKey[TSDB_SUBSCRIBE_KEY_LEN];
SRpcHandleInfo pInfo;
} STqPushEntry;
struct STQ {
SVnode* pVnode;
char* path;
SHashObj* pPushMgr; // consumerId -> STqHandle*
SVnode* pVnode;
char* path;
SRWLatch pushLock;
SHashObj* pPushMgr; // consumerId -> STqPushEntry
SHashObj* pHandle; // subKey -> STqHandle
SHashObj* pCheckInfo; // topic -> SAlterCheckInfo
@ -146,7 +156,9 @@ int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHea
// tqExec
int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp);
int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols);
int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp);
int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry);
// tqMeta
int32_t tqMetaOpen(STQ* pTq);

1
source/dnode/vnode/src/inc/vnd.h
View File

@ -99,7 +99,6 @@ void vnodeSyncStart(SVnode* pVnode);
void vnodeSyncClose(SVnode* pVnode);
void vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg);
bool vnodeIsLeader(SVnode* pVnode);
bool vnodeIsReadyForRead(SVnode* pVnode);
bool vnodeIsRoleLeader(SVnode* pVnode);
#ifdef __cplusplus

115
source/dnode/vnode/src/tq/tq.c
View File

@ -65,6 +65,11 @@ static void destroySTqHandle(void* data) {
}
}
static void tqPushEntryFree(void* data) {
STqPushEntry* p = *(void**)data;
taosMemoryFree(p);
}
STQ* tqOpen(const char* path, SVnode* pVnode) {
STQ* pTq = taosMemoryCalloc(1, sizeof(STQ));
if (pTq == NULL) {
@ -78,7 +83,9 @@ STQ* tqOpen(const char* path, SVnode* pVnode) {
taosHashSetFreeFp(pTq->pHandle, destroySTqHandle);
pTq->pPushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
taosInitRWLatch(&pTq->pushLock);
pTq->pPushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
taosHashSetFreeFp(pTq->pPushMgr, tqPushEntryFree);
pTq->pCheckInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
@ -153,6 +160,65 @@ int32_t tqSendMetaPollRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq,
return 0;
}
int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry) {
SMqDataRsp* pRsp = &pPushEntry->dataRsp;
ASSERT(taosArrayGetSize(pRsp->blockData) == pRsp->blockNum);
ASSERT(taosArrayGetSize(pRsp->blockDataLen) == pRsp->blockNum);
ASSERT(!pRsp->withSchema);
ASSERT(taosArrayGetSize(pRsp->blockSchema) == 0);
if (pRsp->reqOffset.type == TMQ_OFFSET__LOG) {
if (pRsp->blockNum > 0) {
ASSERT(pRsp->rspOffset.version > pRsp->reqOffset.version);
} else {
ASSERT(pRsp->rspOffset.version >= pRsp->reqOffset.version);
}
}
int32_t len = 0;
int32_t code = 0;
tEncodeSize(tEncodeSMqDataRsp, pRsp, len, code);
if (code < 0) {
return -1;
}
int32_t tlen = sizeof(SMqRspHead) + len;
void* buf = rpcMallocCont(tlen);
if (buf == NULL) {
return -1;
}
memcpy(buf, &pPushEntry->rspHead, sizeof(SMqRspHead));
void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
SEncoder encoder = {0};
tEncoderInit(&encoder, abuf, len);
tEncodeSMqDataRsp(&encoder, pRsp);
tEncoderClear(&encoder);
SRpcMsg rsp = {
.info = pPushEntry->pInfo,
.pCont = buf,
.contLen = tlen,
.code = 0,
};
tmsgSendRsp(&rsp);
char buf1[80] = {0};
char buf2[80] = {0};
tFormatOffset(buf1, 80, &pRsp->reqOffset);
tFormatOffset(buf2, 80, &pRsp->rspOffset);
tqDebug("vgId:%d, from consumer:%" PRId64 ", (epoch %d) push rsp, block num: %d, reqOffset:%s, rspOffset:%s",
TD_VID(pTq->pVnode), pPushEntry->rspHead.consumerId, pRsp->head.epoch, pRsp->blockNum, buf1, buf2);
return 0;
}
int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp) {
ASSERT(taosArrayGetSize(pRsp->blockData) == pRsp->blockNum);
ASSERT(taosArrayGetSize(pRsp->blockDataLen) == pRsp->blockNum);
@ -354,6 +420,8 @@ static int32_t tqInitDataRsp(SMqDataRsp* pRsp, const SMqPollReq* pReq, int8_t su
return -1;
}
pRsp->withTbName = 0;
#if 0
pRsp->withTbName = pReq->withTbName;
if (pRsp->withTbName) {
pRsp->blockTbName = taosArrayInit(0, sizeof(void*));
@ -362,6 +430,7 @@ static int32_t tqInitDataRsp(SMqDataRsp* pRsp, const SMqPollReq* pReq, int8_t su
return -1;
}
}
#endif
if (subType == TOPIC_SUB_TYPE__COLUMN) {
pRsp->withSchema = false;
@ -477,11 +546,33 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
SMqDataRsp dataRsp = {0};
tqInitDataRsp(&dataRsp, pReq, pHandle->execHandle.subType);
// lock
taosWLockLatch(&pTq->pushLock);
tqScanData(pTq, pHandle, &dataRsp, &fetchOffsetNew);
#if 1
if (dataRsp.blockNum == 0 && dataRsp.reqOffset.type == TMQ_OFFSET__LOG &&
dataRsp.reqOffset.version == dataRsp.rspOffset.version) {
STqPushEntry* pPushEntry = taosMemoryCalloc(1, sizeof(STqPushEntry));
if (pPushEntry != NULL) {
pPushEntry->pInfo = pMsg->info;
memcpy(pPushEntry->subKey, pHandle->subKey, TSDB_SUBSCRIBE_KEY_LEN);
dataRsp.withTbName = 0;
memcpy(&pPushEntry->dataRsp, &dataRsp, sizeof(SMqDataRsp));
pPushEntry->rspHead.consumerId = consumerId;
pPushEntry->rspHead.epoch = reqEpoch;
pPushEntry->rspHead.mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
taosHashPut(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey) + 1, &pPushEntry, sizeof(void*));
tqDebug("tmq poll: consumer %ld, subkey %s, vg %d save handle to push mgr", consumerId, pHandle->subKey,
TD_VID(pTq->pVnode));
// unlock
taosWUnLockLatch(&pTq->pushLock);
return 0;
}
}
taosWUnLockLatch(&pTq->pushLock);
#endif
if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
code = -1;
}
@ -614,10 +705,22 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
int32_t tqProcessVgDeleteReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg;
int32_t code = taosHashRemove(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
ASSERT(code == 0);
taosWLockLatch(&pTq->pushLock);
int32_t code = taosHashRemove(pTq->pPushMgr, pReq->subKey, strlen(pReq->subKey));
if (code != 0) {
tqDebug("vgId:%d, tq remove push handle %s", pTq->pVnode->config.vgId, pReq->subKey);
}
taosWUnLockLatch(&pTq->pushLock);
tqOffsetDelete(pTq->pOffsetStore, pReq->subKey);
code = taosHashRemove(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
if (code != 0) {
tqError("cannot process tq delete req %s, since no such handle", pReq->subKey);
}
code = tqOffsetDelete(pTq->pOffsetStore, pReq->subKey);
if (code != 0) {
tqError("cannot process tq delete req %s, since no such offset", pReq->subKey);
}
if (tqMetaDeleteHandle(pTq, pReq->subKey) < 0) {
ASSERT(0);
@ -756,7 +859,9 @@ int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLe
atomic_add_fetch_32(&pHandle->epoch, 1);
if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) {
// TODO
ASSERT(0);
}
// close handle
}
return 0;

6
source/dnode/vnode/src/tq/tqExec.c
View File

@ -15,7 +15,7 @@
#include "tq.h"
static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols) {
int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols) {
int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
void* buf = taosMemoryCalloc(1, dataStrLen);
if (buf == NULL) return -1;
@ -243,7 +243,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp
}
if (pHandle->fetchMeta) {
SSubmitBlk* pBlk = pReader->pBlock;
int32_t schemaLen = htonl(pBlk->schemaLen);
int32_t schemaLen = htonl(pBlk->schemaLen);
if (schemaLen > 0) {
if (pRsp->createTableNum == 0) {
pRsp->createTableLen = taosArrayInit(0, sizeof(int32_t));
@ -278,7 +278,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp
}
if (pHandle->fetchMeta) {
SSubmitBlk* pBlk = pReader->pBlock;
int32_t schemaLen = htonl(pBlk->schemaLen);
int32_t schemaLen = htonl(pBlk->schemaLen);
if (schemaLen > 0) {
if (pRsp->createTableNum == 0) {
pRsp->createTableLen = taosArrayInit(0, sizeof(int32_t));

91
source/dnode/vnode/src/tq/tqPush.c
View File

@ -213,6 +213,97 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
#endif
int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
tqDebug("vgId:%d tq push msg ver %ld, type: %s", pTq->pVnode->config.vgId, ver, TMSG_INFO(msgType));
if (msgType == TDMT_VND_SUBMIT) {
// lock push mgr to avoid potential msg lost
taosWLockLatch(&pTq->pushLock);
tqDebug("vgId:%d push handle num %d", pTq->pVnode->config.vgId, taosHashGetSize(pTq->pPushMgr));
if (taosHashGetSize(pTq->pPushMgr) != 0) {
SArray* cachedKeys = taosArrayInit(0, sizeof(void*));
SArray* cachedKeyLens = taosArrayInit(0, sizeof(size_t));
void* data = taosMemoryMalloc(msgLen);
if (data == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tqError("failed to copy data for stream since out of memory");
return -1;
}
memcpy(data, msg, msgLen);
SSubmitReq* pReq = (SSubmitReq*)data;
pReq->version = ver;
void* pIter = NULL;
while (1) {
pIter = taosHashIterate(pTq->pPushMgr, pIter);
if (pIter == NULL) break;
STqPushEntry* pPushEntry = *(STqPushEntry**)pIter;
STqHandle* pHandle = taosHashGet(pTq->pHandle, pPushEntry->subKey, strlen(pPushEntry->subKey));
if (pHandle == NULL) {
tqDebug("vgId:%d cannot find handle %s", pTq->pVnode->config.vgId, pPushEntry->subKey);
continue;
}
if (pPushEntry->dataRsp.reqOffset.version > ver) {
tqDebug("vgId:%d push entry req version %ld, while push version %ld, skip", pTq->pVnode->config.vgId,
pPushEntry->dataRsp.reqOffset.version, ver);
continue;
}
STqExecHandle* pExec = &pHandle->execHandle;
qTaskInfo_t task = pExec->task;
SMqDataRsp* pRsp = &pPushEntry->dataRsp;
// prepare scan mem data
qStreamScanMemData(task, pReq);
// exec
while (1) {
SSDataBlock* pDataBlock = NULL;
uint64_t ts = 0;
if (qExecTask(task, &pDataBlock, &ts) < 0) {
ASSERT(0);
}
if (pDataBlock == NULL) {
break;
}
tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols);
pRsp->blockNum++;
}
tqDebug("vgId:%d tq handle push, subkey: %s, block num: %d", pTq->pVnode->config.vgId, pPushEntry->subKey,
pRsp->blockNum);
if (pRsp->blockNum > 0) {
// set offset
tqOffsetResetToLog(&pRsp->rspOffset, ver);
// remove from hash
size_t kLen;
void* key = taosHashGetKey(pIter, &kLen);
void* keyCopy = taosMemoryMalloc(kLen);
memcpy(keyCopy, key, kLen);
taosArrayPush(cachedKeys, &keyCopy);
taosArrayPush(cachedKeyLens, &kLen);
tqPushDataRsp(pTq, pPushEntry);
}
}
// delete entry
for (int32_t i = 0; i < taosArrayGetSize(cachedKeys); i++) {
void* key = taosArrayGetP(cachedKeys, i);
size_t kLen = *(size_t*)taosArrayGet(cachedKeyLens, i);
if (taosHashRemove(pTq->pPushMgr, key, kLen) != 0) {
ASSERT(0);
}
}
taosArrayDestroyP(cachedKeys, (FDelete)taosMemoryFree);
taosArrayDestroy(cachedKeyLens);
}
// unlock
taosWUnLockLatch(&pTq->pushLock);
}
if (vnodeIsRoleLeader(pTq->pVnode)) {
if (msgType == TDMT_VND_SUBMIT) {
if (taosHashGetSize(pTq->pStreamMeta->pTasks) == 0) return 0;

72
source/dnode/vnode/src/tq/tqRead.c
View File

@ -15,21 +15,20 @@
#include "tq.h"
bool isValValidForTable(STqHandle* pHandle, SWalCont *pHead){
if(pHandle->execHandle.subType != TOPIC_SUB_TYPE__TABLE){
bool isValValidForTable(STqHandle* pHandle, SWalCont* pHead) {
if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__TABLE) {
return true;
}
int16_t msgType = pHead->msgType;
char* body = pHead->body;
int32_t bodyLen = pHead->bodyLen;
int16_t msgType = pHead->msgType;
char* body = pHead->body;
int32_t bodyLen = pHead->bodyLen;
int64_t tbSuid = pHandle->execHandle.execTb.suid;
int64_t realTbSuid = 0;
SDecoder coder;
void* data = POINTER_SHIFT(body, sizeof(SMsgHead));
int32_t len = bodyLen - sizeof(SMsgHead);
int64_t tbSuid = pHandle->execHandle.execTb.suid;
int64_t realTbSuid = 0;
SDecoder coder;
void* data = POINTER_SHIFT(body, sizeof(SMsgHead));
int32_t len = bodyLen - sizeof(SMsgHead);
tDecoderInit(&coder, data, len);
if (msgType == TDMT_VND_CREATE_STB || msgType == TDMT_VND_ALTER_STB) {
@ -43,38 +42,38 @@ bool isValValidForTable(STqHandle* pHandle, SWalCont *pHead){
if (tDecodeSVDropStbReq(&coder, &req) < 0) {
goto end;
}
realTbSuid = req.suid;
realTbSuid = req.suid;
} else if (msgType == TDMT_VND_CREATE_TABLE) {
SVCreateTbBatchReq req = {0};
if (tDecodeSVCreateTbBatchReq(&coder, &req) < 0) {
goto end;
}
int32_t needRebuild = 0;
int32_t needRebuild = 0;
SVCreateTbReq* pCreateReq = NULL;
for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
pCreateReq = req.pReqs + iReq;
if(pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid){
if (pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid) {
needRebuild++;
}
}
if(needRebuild == 0){
if (needRebuild == 0) {
// do nothing
}else if(needRebuild == req.nReqs){
} else if (needRebuild == req.nReqs) {
realTbSuid = tbSuid;
}else{
} else {
realTbSuid = tbSuid;
SVCreateTbBatchReq reqNew = {0};
reqNew.pArray = taosArrayInit(req.nReqs, sizeof(struct SVCreateTbReq));
for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
pCreateReq = req.pReqs + iReq;
if(pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid){
if (pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid) {
reqNew.nReqs++;
taosArrayPush(reqNew.pArray, pCreateReq);
}
}
int tlen;
int tlen;
int32_t ret = 0;
tEncodeSize(tEncodeSVCreateTbBatchReq, &reqNew, tlen, ret);
void* buf = taosMemoryMalloc(tlen);
@ -107,7 +106,7 @@ bool isValValidForTable(STqHandle* pHandle, SWalCont *pHead){
}
}
} else if (msgType == TDMT_VND_ALTER_TABLE) {
SVAlterTbReq req = {0};
SVAlterTbReq req = {0};
if (tDecodeSVAlterTbReq(&coder, &req) < 0) {
goto end;
@ -129,32 +128,32 @@ bool isValValidForTable(STqHandle* pHandle, SWalCont *pHead){
goto end;
}
int32_t needRebuild = 0;
int32_t needRebuild = 0;
SVDropTbReq* pDropReq = NULL;
for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
pDropReq = req.pReqs + iReq;
if(pDropReq->suid == tbSuid){
if (pDropReq->suid == tbSuid) {
needRebuild++;
}
}
if(needRebuild == 0){
if (needRebuild == 0) {
// do nothing
}else if(needRebuild == req.nReqs){
} else if (needRebuild == req.nReqs) {
realTbSuid = tbSuid;
}else{
} else {
realTbSuid = tbSuid;
SVDropTbBatchReq reqNew = {0};
reqNew.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbReq));
for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
pDropReq = req.pReqs + iReq;
if(pDropReq->suid == tbSuid){
if (pDropReq->suid == tbSuid) {
reqNew.nReqs++;
taosArrayPush(reqNew.pArray, pDropReq);
}
}
int tlen;
int tlen;
int32_t ret = 0;
tEncodeSize(tEncodeSVDropTbBatchReq, &reqNew, tlen, ret);
void* buf = taosMemoryMalloc(tlen);
@ -177,11 +176,11 @@ bool isValValidForTable(STqHandle* pHandle, SWalCont *pHead){
goto end;
}
realTbSuid = req.suid;
} else{
} else {
ASSERT(0);
}
end:
end:
tDecoderClear(&coder);
return tbSuid == realTbSuid;
}
@ -224,7 +223,7 @@ int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHea
code = -1;
goto END;
}
if(isValValidForTable(pHandle, pHead)){
if (isValValidForTable(pHandle, pHead)) {
*fetchOffset = offset;
code = 0;
goto END;
@ -241,7 +240,7 @@ int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHea
offset++;
}
}
END:
END:
taosThreadMutexUnlock(&pHandle->pWalReader->mutex);
return code;
}
@ -315,14 +314,18 @@ int32_t tqNextBlock(STqReader* pReader, SFetchRet* ret) {
return -1;
}
void* body = pReader->pWalReader->pHead->head.body;
#if 0
if (pReader->pWalReader->pHead->head.msgType != TDMT_VND_SUBMIT) {
// TODO do filter
ret->fetchType = FETCH_TYPE__META;
ret->meta = pReader->pWalReader->pHead->head.body;
return 0;
} else {
tqReaderSetDataMsg(pReader, body, pReader->pWalReader->pHead->head.version);
#endif
tqReaderSetDataMsg(pReader, body, pReader->pWalReader->pHead->head.version);
#if 0
}
#endif
}
while (tqNextDataBlock(pReader)) {
@ -334,6 +337,7 @@ int32_t tqNextBlock(STqReader* pReader, SFetchRet* ret) {
continue;
}
ret->fetchType = FETCH_TYPE__DATA;
tqDebug("return data rows %d", ret->data.info.rows);
return 0;
}
@ -341,14 +345,14 @@ int32_t tqNextBlock(STqReader* pReader, SFetchRet* ret) {
ret->offset.type = TMQ_OFFSET__LOG;
ret->offset.version = pReader->ver;
ASSERT(pReader->ver >= 0);
ret->fetchType = FETCH_TYPE__NONE;
ret->fetchType = FETCH_TYPE__SEP;
tqDebug("return offset %" PRId64 ", processed finish", ret->offset.version);
return 0;
}
}
}
int32_t tqReaderSetDataMsg(STqReader* pReader, SSubmitReq* pMsg, int64_t ver) {
int32_t tqReaderSetDataMsg(STqReader* pReader, const SSubmitReq* pMsg, int64_t ver) {
pReader->pMsg = pMsg;
if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1;

26
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -486,7 +486,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
pReader->suid = pCond->suid;
pReader->order = pCond->order;
pReader->capacity = 4096;
pReader->capacity = capacity;
pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL;
pReader->verRange = getQueryVerRange(pVnode, pCond, level);
pReader->type = pCond->type;
@ -841,14 +841,18 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
bool asc = ASCENDING_TRAVERSE(pReader->order);
int32_t step = asc ? 1 : -1;
if (asc && pReader->window.skey <= pBlock->minKey.ts) {
pDumpInfo->rowIndex = 0;
} else if (!asc && pReader->window.ekey >= pBlock->maxKey.ts) {
pDumpInfo->rowIndex = pBlock->nRow - 1;
} else {
int32_t pos = asc ? pBlock->nRow - 1 : 0;
int32_t order = (pReader->order == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, pReader->window.skey, order);
if ((pDumpInfo->rowIndex == 0 && asc) || (pDumpInfo->rowIndex == pBlock->nRow - 1 && (!asc))) {
if (asc && pReader->window.skey <= pBlock->minKey.ts) {
//pDumpInfo->rowIndex = 0;
} else
if (!asc && pReader->window.ekey >= pBlock->maxKey.ts) {
//pDumpInfo->rowIndex = pBlock->nRow - 1;
} else {
int32_t pos = asc ? pBlock->nRow - 1 : 0;
int32_t order = (pReader->order == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC;
pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, pReader->window.skey, order);
}
}
// time window check
@ -932,8 +936,8 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
pDumpInfo->rowIndex += step * remain;
if (pDumpInfo->rowIndex >= 0 && pDumpInfo->rowIndex < pBlock->nRow) {
int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
setBlockAllDumped(pDumpInfo, ts, pReader->order);
// int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
// setBlockAllDumped(pDumpInfo, ts, pReader->order);
} else {
int64_t k = asc ? pBlock->maxKey.ts : pBlock->minKey.ts;
setBlockAllDumped(pDumpInfo, k, pReader->order);

13
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -289,7 +289,7 @@ int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
vTrace("message in vnode query queue is processing");
if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsReadyForRead(pVnode)) {
if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsLeader(pVnode)) {
vnodeRedirectRpcMsg(pVnode, pMsg);
return 0;
}
@ -311,7 +311,12 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg);
if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||
pMsg->msgType == TDMT_VND_BATCH_META) &&
!vnodeIsReadyForRead(pVnode)) {
!vnodeIsLeader(pVnode)) {
vnodeRedirectRpcMsg(pVnode, pMsg);
return 0;
}
if (pMsg->msgType == TDMT_VND_CONSUME && !pVnode->restored) {
vnodeRedirectRpcMsg(pVnode, pMsg);
return 0;
}
@ -808,7 +813,6 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
SSubmitRsp submitRsp = {0};
SSubmitMsgIter msgIter = {0};
SSubmitBlk *pBlock;
SSubmitRsp rsp = {0};
SVCreateTbReq createTbReq = {0};
SDecoder decoder = {0};
int32_t nRows;
@ -921,7 +925,8 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
}
if (taosArrayGetSize(newTbUids) > 0) {
vDebug("vgId:%d, add %d table into query table list in handling submit", TD_VID(pVnode), (int32_t)taosArrayGetSize(newTbUids));
vDebug("vgId:%d, add %d table into query table list in handling submit", TD_VID(pVnode),
(int32_t)taosArrayGetSize(newTbUids));
}
tqUpdateTbUidList(pVnode->pTq, newTbUids, true);

23
source/dnode/vnode/src/vnd/vnodeSync.c
View File

@ -240,7 +240,7 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
isWeak, isBlock, msg, numOfMsgs, arrayPos, pMsg->info.handle);
if (!pVnode->restored) {
vGError("vgId:%d, msg:%p failed to process since not leader", vgId, pMsg);
vGError("vgId:%d, msg:%p failed to process since restore not finished", vgId, pMsg);
terrno = TSDB_CODE_APP_NOT_READY;
vnodeHandleProposeError(pVnode, pMsg, TSDB_CODE_APP_NOT_READY);
rpcFreeCont(pMsg->pCont);
@ -496,16 +496,16 @@ static int32_t vnodeSyncGetSnapshot(SSyncFSM *pFsm, SSnapshot *pSnapshot) {
return 0;
}
static void vnodeSyncReconfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta cbMeta) {
static void vnodeSyncReconfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta *cbMeta) {
SVnode *pVnode = pFsm->data;
SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
syncGetAndDelRespRpc(pVnode->sync, cbMeta.newCfgSeqNum, &rpcMsg.info);
rpcMsg.info.conn.applyIndex = cbMeta.index;
syncGetAndDelRespRpc(pVnode->sync, cbMeta->newCfgSeqNum, &rpcMsg.info);
rpcMsg.info.conn.applyIndex = cbMeta->index;
const STraceId *trace = (STraceId *)&pMsg->info.traceId;
vGTrace("vgId:%d, alter vnode replica is confirmed, type:%s contLen:%d seq:%" PRIu64 " handle:%p", TD_VID(pVnode),
TMSG_INFO(pMsg->msgType), pMsg->contLen, cbMeta.seqNum, rpcMsg.info.handle);
TMSG_INFO(pMsg->msgType), pMsg->contLen, cbMeta->seqNum, rpcMsg.info.handle);
if (rpcMsg.info.handle != NULL) {
tmsgSendRsp(&rpcMsg);
}
@ -796,16 +796,3 @@ bool vnodeIsLeader(SVnode *pVnode) {
return true;
}
bool vnodeIsReadyForRead(SVnode *pVnode) {
if (syncIsReady(pVnode->sync)) {
return true;
}
if (syncIsReadyForRead(pVnode->sync)) {
return true;
}
vDebug("vgId:%d, vnode not ready for read, state:%s, last:%ld, cmt:%ld", pVnode->config.vgId,
syncGetMyRoleStr(pVnode->sync), syncGetLastIndex(pVnode->sync), syncGetCommitIndex(pVnode->sync));
return false;
}

8
source/libs/catalog/src/ctgAsync.c
View File

@ -487,6 +487,8 @@ int32_t ctgInitTask(SCtgJob* pJob, CTG_TASK_TYPE type, void* param, int32_t* tas
int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const SCatalogReq* pReq, catalogCallback fp,
void* param) {
int32_t code = 0;
int64_t st = taosGetTimestampUs();
int32_t tbMetaNum = (int32_t)ctgGetTablesReqNum(pReq->pTableMeta);
int32_t dbVgNum = (int32_t)taosArrayGetSize(pReq->pDbVgroup);
int32_t tbHashNum = (int32_t)ctgGetTablesReqNum(pReq->pTableHash);
@ -634,12 +636,12 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const
taosAcquireRef(gCtgMgmt.jobPool, pJob->refId);
qDebug("QID:0x%" PRIx64 ", jobId: 0x%" PRIx64 " initialized, task num %d, forceUpdate %d", pJob->queryId, pJob->refId,
taskNum, pReq->forceUpdate);
double el = (taosGetTimestampUs() - st)/1000.0;
qDebug("QID:0x%" PRIx64 ", jobId: 0x%" PRIx64 " initialized, task num %d, forceUpdate %d, elapsed time:%.2f ms",
pJob->queryId, pJob->refId, taskNum, pReq->forceUpdate, el);
return TSDB_CODE_SUCCESS;
_return:
ctgFreeJob(*job);
CTG_RET(code);
}

2
source/libs/executor/inc/executorimpl.h
View File

@ -146,6 +146,7 @@ typedef struct {
SMqMetaRsp metaRsp; // for tmq fetching meta
int8_t returned;
int64_t snapshotVer;
const SSubmitReq* pReq;
SSchemaWrapper* schema;
char tbName[TSDB_TABLE_NAME_LEN];
@ -192,6 +193,7 @@ enum {
OP_OPENED = 0x1,
OP_RES_TO_RETURN = 0x5,
OP_EXEC_DONE = 0x9,
OP_EXEC_RECV = 0x11,
};
typedef struct SOperatorFpSet {

10
source/libs/executor/src/executor.c
View File

@ -486,7 +486,7 @@ int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bo
if (pLocal) {
memcpy(&pTaskInfo->localFetch, pLocal, sizeof(*pLocal));
}
taosArrayClearEx(pResList, freeBlock);
int64_t curOwner = 0;
@ -774,6 +774,14 @@ int32_t initQueryTableDataCondForTmq(SQueryTableDataCond* pCond, SSnapContext* s
return TSDB_CODE_SUCCESS;
}
int32_t qStreamScanMemData(qTaskInfo_t tinfo, const SSubmitReq* pReq) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
ASSERT(pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE);
ASSERT(pTaskInfo->streamInfo.pReq == NULL);
pTaskInfo->streamInfo.pReq = pReq;
return 0;
}
int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
SOperatorInfo* pOperator = pTaskInfo->pRoot;

8
source/libs/executor/src/executorimpl.c
View File

@ -1273,6 +1273,14 @@ static void doCopyResultToDataBlock(SExprInfo* pExprInfo, int32_t numOfExprs, SR
pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowEntryOffset);
if (pCtx[j].fpSet.finalize) {
if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_group_key") == 0) {
// for groupkey along with functions that output multiple lines(e.g. Histogram)
// need to match groupkey result for each output row of that function.
if (pCtx[j].resultInfo->numOfRes != 0) {
pCtx[j].resultInfo->numOfRes = pRow->numOfRows;
}
}
int32_t code = pCtx[j].fpSet.finalize(&pCtx[j], pBlock);
if (TAOS_FAILED(code)) {
qError("%s build result data block error, code %s", GET_TASKID(pTaskInfo), tstrerror(code));

26
source/libs/executor/src/projectoperator.c
View File

@ -53,7 +53,7 @@ static void destroyIndefinitOperatorInfo(void* param) {
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode,
SExecTaskInfo* pTaskInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t code = TSDB_CODE_SUCCESS;
SProjectOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SProjectOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
@ -184,7 +184,7 @@ static int32_t doIngroupLimitOffset(SLimitInfo* pLimitInfo, uint64_t groupId, SS
if (pLimitInfo->limit.limit >= 0 && pLimitInfo->numOfOutputRows + pBlock->info.rows >= pLimitInfo->limit.limit) {
int32_t keepRows = (int32_t)(pLimitInfo->limit.limit - pLimitInfo->numOfOutputRows);
blockDataKeepFirstNRows(pBlock, keepRows);
//TODO: optimize it later when partition by + limit
// TODO: optimize it later when partition by + limit
if ((pLimitInfo->slimit.limit == -1 && pLimitInfo->currentGroupId == 0) ||
(pLimitInfo->slimit.limit > 0 && pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
doSetOperatorCompleted(pOperator);
@ -206,9 +206,16 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
blockDataCleanup(pFinalRes);
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (pTaskInfo->streamInfo.pReq) {
pOperator->status = OP_OPENED;
}
qDebug("enter project");
if (pOperator->status == OP_EXEC_DONE) {
if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
pOperator->status = OP_OPENED;
qDebug("projection in queue model, set status open and return null");
return NULL;
}
@ -237,9 +244,23 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
// The downstream exec may change the value of the newgroup, so use a local variable instead.
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL) {
if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE && pFinalRes->info.rows == 0) {
pOperator->status = OP_OPENED;
if (pOperator->status == OP_EXEC_RECV) {
continue;
} else {
return NULL;
}
}
qDebug("set op close, exec %d, status %d rows %d", pTaskInfo->execModel, pOperator->status,
pFinalRes->info.rows);
doSetOperatorCompleted(pOperator);
break;
}
if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
qDebug("set status recv");
pOperator->status = OP_EXEC_RECV;
}
// for stream interval
if (pBlock->info.type == STREAM_RETRIEVE || pBlock->info.type == STREAM_DELETE_RESULT ||
@ -298,6 +319,7 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
// when apply the limit/offset for each group, pRes->info.rows may be 0, due to limit constraint.
if (pFinalRes->info.rows > 0 || (pOperator->status == OP_EXEC_DONE)) {
qDebug("project return %d rows, status %d", pFinalRes->info.rows, pOperator->status);
break;
}
} else {

338
source/libs/executor/src/scanoperator.c
View File

@ -41,7 +41,12 @@ static int32_t buildDbTableInfoBlock(bool sysInfo, const SSDataBlock* p, const S
static bool processBlockWithProbability(const SSampleExecInfo* pInfo);
bool processBlockWithProbability(const SSampleExecInfo* pInfo) {
static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo, SMetaReader* smr, const char* dbname,
const char* tableName, int32_t* pNumOfRows,
const SSDataBlock* dataBlock);
static void relocateAndFilterSysTagsScanResult(SSysTableScanInfo* pInfo, int32_t numOfRows, SSDataBlock* dataBlock);
bool processBlockWithProbability(const SSampleExecInfo* pInfo) {
#if 0
if (pInfo->sampleRatio == 1) {
return true;
@ -1435,6 +1440,43 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
SStreamScanInfo* pInfo = pOperator->info;
qDebug("queue scan called");
if (pTaskInfo->streamInfo.pReq != NULL) {
if (pInfo->tqReader->pMsg == NULL) {
pInfo->tqReader->pMsg = pTaskInfo->streamInfo.pReq;
const SSubmitReq* pSubmit = pInfo->tqReader->pMsg;
if (tqReaderSetDataMsg(pInfo->tqReader, pSubmit, 0) < 0) {
qError("submit msg messed up when initing stream submit block %p", pSubmit);
pInfo->tqReader->pMsg = NULL;
pTaskInfo->streamInfo.pReq = NULL;
ASSERT(0);
}
}
blockDataCleanup(pInfo->pRes);
SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
while (tqNextDataBlock(pInfo->tqReader)) {
SSDataBlock block = {0};
int32_t code = tqRetrieveDataBlock(&block, pInfo->tqReader);
if (code != TSDB_CODE_SUCCESS || block.info.rows == 0) {
continue;
}
setBlockIntoRes(pInfo, &block);
if (pBlockInfo->rows > 0) {
return pInfo->pRes;
}
}
pInfo->tqReader->pMsg = NULL;
pTaskInfo->streamInfo.pReq = NULL;
return NULL;
}
if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
if (pResult && pResult->info.rows > 0) {
@ -1467,8 +1509,8 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
if (setBlockIntoRes(pInfo, &ret.data) < 0) {
ASSERT(0);
}
// TODO clean data block
if (pInfo->pRes->info.rows > 0) {
pOperator->status = OP_EXEC_RECV;
qDebug("queue scan log return %d rows", pInfo->pRes->info.rows);
return pInfo->pRes;
}
@ -1477,18 +1519,19 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
// pTaskInfo->streamInfo.lastStatus = ret.offset;
// pTaskInfo->streamInfo.metaBlk = ret.meta;
// return NULL;
} else if (ret.fetchType == FETCH_TYPE__NONE) {
} else if (ret.fetchType == FETCH_TYPE__NONE ||
(ret.fetchType == FETCH_TYPE__SEP && pOperator->status == OP_EXEC_RECV)) {
pTaskInfo->streamInfo.lastStatus = ret.offset;
ASSERT(pTaskInfo->streamInfo.lastStatus.version >= pTaskInfo->streamInfo.prepareStatus.version);
ASSERT(pTaskInfo->streamInfo.lastStatus.version + 1 == pInfo->tqReader->pWalReader->curVersion);
char formatBuf[80];
tFormatOffset(formatBuf, 80, &ret.offset);
qDebug("queue scan log return null, offset %s", formatBuf);
pOperator->status = OP_OPENED;
return NULL;
} else {
ASSERT(0);
}
}
#if 0
} else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
if (pResult && pResult->info.rows > 0) {
@ -1497,6 +1540,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
}
qDebug("stream scan tsdb return null");
return NULL;
#endif
} else {
ASSERT(0);
return NULL;
@ -2360,6 +2404,48 @@ int32_t convertTagDataToStr(char* str, int type, void* buf, int32_t bufSize, int
return TSDB_CODE_SUCCESS;
}
static bool sysTableIsOperatorCondOnOneTable(SNode* pCond, char* condTable) {
SOperatorNode* node = (SOperatorNode*)pCond;
if (node->opType == OP_TYPE_EQUAL) {
if (nodeType(node->pLeft) == QUERY_NODE_COLUMN &&
strcasecmp(nodesGetNameFromColumnNode(node->pLeft), "table_name") == 0 &&
nodeType(node->pRight) == QUERY_NODE_VALUE) {
SValueNode* pValue = (SValueNode*)node->pRight;
if (pValue->node.type == TSDB_DATA_TYPE_NCHAR || pValue->node.type == TSDB_DATA_TYPE_VARCHAR ||
pValue->node.type == TSDB_DATA_TYPE_BINARY) {
char* value = nodesGetStrValueFromNode(pValue);
strncpy(condTable, value, TSDB_TABLE_NAME_LEN);
taosMemoryFree(value);
return true;
}
}
}
return false;
}
static bool sysTableIsCondOnOneTable(SNode* pCond, char* condTable) {
if (pCond == NULL) {
return false;
}
if (nodeType(pCond) == QUERY_NODE_LOGIC_CONDITION) {
SLogicConditionNode* node = (SLogicConditionNode*)pCond;
if (LOGIC_COND_TYPE_AND == node->condType) {
SNode* pChild = NULL;
FOREACH(pChild, node->pParameterList) {
if (QUERY_NODE_OPERATOR == nodeType(pChild) && sysTableIsOperatorCondOnOneTable(pChild, condTable)) {
return true;
}
}
}
}
if (QUERY_NODE_OPERATOR == nodeType(pCond)) {
return sysTableIsOperatorCondOnOneTable(pCond, condTable);
}
return false;
}
static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SSysTableScanInfo* pInfo = pOperator->info;
@ -2367,13 +2453,12 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
return NULL;
}
if (pInfo->pCur == NULL) {
pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
}
blockDataCleanup(pInfo->pRes);
int32_t numOfRows = 0;
SSDataBlock* dataBlock = buildInfoSchemaTableMetaBlock(TSDB_INS_TABLE_TAGS);
blockDataEnsureCapacity(dataBlock, pOperator->resultInfo.capacity);
const char* db = NULL;
int32_t vgId = 0;
vnodeGetInfo(pInfo->readHandle.vnode, &db, &vgId);
@ -2385,10 +2470,32 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
tNameGetDbName(&sn, varDataVal(dbname));
varDataSetLen(dbname, strlen(varDataVal(dbname)));
SSDataBlock* p = buildInfoSchemaTableMetaBlock(TSDB_INS_TABLE_TAGS);
blockDataEnsureCapacity(p, pOperator->resultInfo.capacity);
char condTableName[TSDB_TABLE_NAME_LEN] = {0};
// optimize when sql like where table_name='tablename' and xxx.
if (pInfo->pCondition && sysTableIsCondOnOneTable(pInfo->pCondition, condTableName)) {
char tableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(tableName, condTableName);
SMetaReader smr = {0};
metaReaderInit(&smr, pInfo->readHandle.meta, 0);
metaGetTableEntryByName(&smr, condTableName);
sysTableUserTagsFillOneTableTags(pInfo, &smr, dbname, tableName, &numOfRows, dataBlock);
metaReaderClear(&smr);
if (numOfRows > 0) {
relocateAndFilterSysTagsScanResult(pInfo, numOfRows, dataBlock);
numOfRows = 0;
}
blockDataDestroy(dataBlock);
pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
}
int32_t ret = 0;
if (pInfo->pCur == NULL) {
pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
}
while ((ret = metaTbCursorNext(pInfo->pCur)) == 0) {
if (pInfo->pCur->mr.me.type != TSDB_CHILD_TABLE) {
continue;
@ -2411,100 +2518,12 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
char stableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(stableName, smr.me.name);
sysTableUserTagsFillOneTableTags(pInfo, &smr, dbname, tableName, &numOfRows, dataBlock);
int32_t numOfTags = smr.me.stbEntry.schemaTag.nCols;
for (int32_t i = 0; i < numOfTags; ++i) {
SColumnInfoData* pColInfoData = NULL;
// table name
pColInfoData = taosArrayGet(p->pDataBlock, 0);
colDataAppend(pColInfoData, numOfRows, tableName, false);
// database name
pColInfoData = taosArrayGet(p->pDataBlock, 1);
colDataAppend(pColInfoData, numOfRows, dbname, false);
// super table name
pColInfoData = taosArrayGet(p->pDataBlock, 2);
colDataAppend(pColInfoData, numOfRows, stableName, false);
// tag name
char tagName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(tagName, smr.me.stbEntry.schemaTag.pSchema[i].name);
pColInfoData = taosArrayGet(p->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, tagName, false);
// tag type
int8_t tagType = smr.me.stbEntry.schemaTag.pSchema[i].type;
pColInfoData = taosArrayGet(p->pDataBlock, 4);
char tagTypeStr[VARSTR_HEADER_SIZE + 32];
int tagTypeLen = sprintf(varDataVal(tagTypeStr), "%s", tDataTypes[tagType].name);
if (tagType == TSDB_DATA_TYPE_VARCHAR) {
tagTypeLen += sprintf(varDataVal(tagTypeStr) + tagTypeLen, "(%d)",
(int32_t)(smr.me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE));
} else if (tagType == TSDB_DATA_TYPE_NCHAR) {
tagTypeLen +=
sprintf(varDataVal(tagTypeStr) + tagTypeLen, "(%d)",
(int32_t)((smr.me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
}
varDataSetLen(tagTypeStr, tagTypeLen);
colDataAppend(pColInfoData, numOfRows, (char*)tagTypeStr, false);
STagVal tagVal = {0};
tagVal.cid = smr.me.stbEntry.schemaTag.pSchema[i].colId;
char* tagData = NULL;
uint32_t tagLen = 0;
if (tagType == TSDB_DATA_TYPE_JSON) {
tagData = (char*)pInfo->pCur->mr.me.ctbEntry.pTags;
} else {
bool exist = tTagGet((STag*)pInfo->pCur->mr.me.ctbEntry.pTags, &tagVal);
if (exist) {
if (IS_VAR_DATA_TYPE(tagType)) {
tagData = (char*)tagVal.pData;
tagLen = tagVal.nData;
} else {
tagData = (char*)&tagVal.i64;
tagLen = tDataTypes[tagType].bytes;
}
}
}
char* tagVarChar = NULL;
if (tagData != NULL) {
if (tagType == TSDB_DATA_TYPE_JSON) {
char* tagJson = parseTagDatatoJson(tagData);
tagVarChar = taosMemoryMalloc(strlen(tagJson) + VARSTR_HEADER_SIZE);
memcpy(varDataVal(tagVarChar), tagJson, strlen(tagJson));
varDataSetLen(tagVarChar, strlen(tagJson));
taosMemoryFree(tagJson);
} else {
int32_t bufSize = IS_VAR_DATA_TYPE(tagType) ? (tagLen + VARSTR_HEADER_SIZE)
: (3 + DBL_MANT_DIG - DBL_MIN_EXP + VARSTR_HEADER_SIZE);
tagVarChar = taosMemoryMalloc(bufSize);
int32_t len = -1;
convertTagDataToStr(varDataVal(tagVarChar), tagType, tagData, tagLen, &len);
varDataSetLen(tagVarChar, len);
}
}
pColInfoData = taosArrayGet(p->pDataBlock, 5);
colDataAppend(pColInfoData, numOfRows, tagVarChar,
(tagData == NULL) || (tagType == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(tagData)));
taosMemoryFree(tagVarChar);
++numOfRows;
}
metaReaderClear(&smr);
if (numOfRows >= pOperator->resultInfo.capacity) {
p->info.rows = numOfRows;
pInfo->pRes->info.rows = numOfRows;
relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
doFilterResult(pInfo);
blockDataCleanup(p);
relocateAndFilterSysTagsScanResult(pInfo, numOfRows, dataBlock);
numOfRows = 0;
if (pInfo->pRes->info.rows > 0) {
@ -2514,19 +2533,11 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
}
if (numOfRows > 0) {
p->info.rows = numOfRows;
pInfo->pRes->info.rows = numOfRows;
relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
doFilterResult(pInfo);
blockDataCleanup(p);
relocateAndFilterSysTagsScanResult(pInfo, numOfRows, dataBlock);
numOfRows = 0;
}
blockDataDestroy(p);
// todo temporarily free the cursor here, the true reason why the free is not valid needs to be found
blockDataDestroy(dataBlock);
if (ret != 0) {
metaCloseTbCursor(pInfo->pCur);
pInfo->pCur = NULL;
@ -2537,6 +2548,111 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
}
static void relocateAndFilterSysTagsScanResult(SSysTableScanInfo* pInfo, int32_t numOfRows, SSDataBlock* dataBlock) {
dataBlock->info.rows = numOfRows;
pInfo->pRes->info.rows = numOfRows;
relocateColumnData(pInfo->pRes, pInfo->scanCols, dataBlock->pDataBlock, false);
doFilterResult(pInfo);
blockDataCleanup(dataBlock);
}
static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo, SMetaReader* smr, const char* dbname,
const char* tableName, int32_t* pNumOfRows,
const SSDataBlock* dataBlock) {
char stableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(stableName, (*smr).me.name);
int32_t numOfRows = *pNumOfRows;
int32_t numOfTags = (*smr).me.stbEntry.schemaTag.nCols;
for (int32_t i = 0; i < numOfTags; ++i) {
SColumnInfoData* pColInfoData = NULL;
// table name
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 0);
colDataAppend(pColInfoData, numOfRows, tableName, false);
// database name
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 1);
colDataAppend(pColInfoData, numOfRows, dbname, false);
// super table name
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 2);
colDataAppend(pColInfoData, numOfRows, stableName, false);
// tag name
char tagName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(tagName, (*smr).me.stbEntry.schemaTag.pSchema[i].name);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, tagName, false);
// tag type
int8_t tagType = (*smr).me.stbEntry.schemaTag.pSchema[i].type;
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 4);
char tagTypeStr[VARSTR_HEADER_SIZE + 32];
int tagTypeLen = sprintf(varDataVal(tagTypeStr), "%s", tDataTypes[tagType].name);
if (tagType == TSDB_DATA_TYPE_VARCHAR) {
tagTypeLen += sprintf(varDataVal(tagTypeStr) + tagTypeLen, "(%d)",
(int32_t)((*smr).me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE));
} else if (tagType == TSDB_DATA_TYPE_NCHAR) {
tagTypeLen +=
sprintf(varDataVal(tagTypeStr) + tagTypeLen, "(%d)",
(int32_t)(((*smr).me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
}
varDataSetLen(tagTypeStr, tagTypeLen);
colDataAppend(pColInfoData, numOfRows, (char*)tagTypeStr, false);
STagVal tagVal = {0};
tagVal.cid = (*smr).me.stbEntry.schemaTag.pSchema[i].colId;
char* tagData = NULL;
uint32_t tagLen = 0;
if (tagType == TSDB_DATA_TYPE_JSON) {
tagData = (char*)pInfo->pCur->mr.me.ctbEntry.pTags;
} else {
bool exist = tTagGet((STag*)pInfo->pCur->mr.me.ctbEntry.pTags, &tagVal);
if (exist) {
if (IS_VAR_DATA_TYPE(tagType)) {
tagData = (char*)tagVal.pData;
tagLen = tagVal.nData;
} else {
tagData = (char*)&tagVal.i64;
tagLen = tDataTypes[tagType].bytes;
}
}
}
char* tagVarChar = NULL;
if (tagData != NULL) {
if (tagType == TSDB_DATA_TYPE_JSON) {
char* tagJson = parseTagDatatoJson(tagData);
tagVarChar = taosMemoryMalloc(strlen(tagJson) + VARSTR_HEADER_SIZE);
memcpy(varDataVal(tagVarChar), tagJson, strlen(tagJson));
varDataSetLen(tagVarChar, strlen(tagJson));
taosMemoryFree(tagJson);
} else {
int32_t bufSize = IS_VAR_DATA_TYPE(tagType) ? (tagLen + VARSTR_HEADER_SIZE)
: (3 + DBL_MANT_DIG - DBL_MIN_EXP + VARSTR_HEADER_SIZE);
tagVarChar = taosMemoryMalloc(bufSize);
int32_t len = -1;
convertTagDataToStr(varDataVal(tagVarChar), tagType, tagData, tagLen, &len);
varDataSetLen(tagVarChar, len);
}
}
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 5);
colDataAppend(pColInfoData, numOfRows, tagVarChar,
(tagData == NULL) || (tagType == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(tagData)));
taosMemoryFree(tagVarChar);
++numOfRows;
}
*pNumOfRows = numOfRows;
return TSDB_CODE_SUCCESS;
}
static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SSysTableScanInfo* pInfo = pOperator->info;

15
source/libs/function/src/builtins.c
View File

@ -2020,6 +2020,11 @@ static int32_t translateUserFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t le
return TSDB_CODE_SUCCESS;
}
static int32_t translateTagsPseudoColumn(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
// The _tags pseudo-column will be expanded to the actual tags on the client side
return TSDB_CODE_SUCCESS;
}
// clang-format off
const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
@ -3156,6 +3161,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_tags",
.type = FUNCTION_TYPE_TAGS,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_MULTI_RES_FUNC,
.translateFunc = translateTagsPseudoColumn,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
};
// clang-format on

5
source/libs/function/src/builtinsimpl.c
View File

@ -6157,7 +6157,10 @@ int32_t groupKeyFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SGroupKeyInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
if (pInfo->hasResult) {
colDataAppend(pCol, pBlock->info.rows, pInfo->data, pInfo->isNull ? true : false);
int32_t currentRow = pBlock->info.rows;
for (; currentRow < pBlock->info.rows + pResInfo->numOfRes; ++currentRow) {
colDataAppend(pCol, currentRow, pInfo->data, pInfo->isNull ? true : false);
}
} else {
pResInfo->numOfRes = 0;
}

2
source/libs/function/src/tudf.c
View File

@ -1656,6 +1656,8 @@ int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
int32_t errCode = udfcRunUdfUvTask(task, UV_TASK_CONNECT);
if (errCode != 0) {
fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfdProxy)->udfdPipeName);
taosMemoryFree(task->session);
taosMemoryFree(task);
return TSDB_CODE_UDF_PIPE_CONNECT_ERR;
}

2
source/libs/function/src/udfd.c
View File

@ -960,7 +960,7 @@ int32_t udfdInitResidentFuncs() {
char* token;
while ((token = strtok_r(pSave, ",", &pSave)) != NULL) {
char func[TSDB_FUNC_NAME_LEN] = {0};
strncpy(func, token, strlen(token));
strncpy(func, token, sizeof(func));
taosArrayPush(global.residentFuncs, func);
}

4
source/libs/nodes/src/nodesMsgFuncs.c
View File

@ -90,12 +90,12 @@ static void endTlvEncode(STlvEncoder* pEncoder, char** pMsg, int32_t* pLen) {
static int32_t tlvEncodeImpl(STlvEncoder* pEncoder, int16_t type, const void* pValue, int32_t len) {
int32_t tlvLen = sizeof(STlv) + len;
if (pEncoder->offset + tlvLen > pEncoder->allocSize) {
void* pNewBuf = taosMemoryRealloc(pEncoder->pBuf, pEncoder->allocSize * 2);
pEncoder->allocSize = TMAX(pEncoder->allocSize * 2, pEncoder->allocSize + pEncoder->offset + tlvLen);
void* pNewBuf = taosMemoryRealloc(pEncoder->pBuf, pEncoder->allocSize);
if (NULL == pNewBuf) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pEncoder->pBuf = pNewBuf;
pEncoder->allocSize = pEncoder->allocSize * 2;
}
STlv* pTlv = (STlv*)(pEncoder->pBuf + pEncoder->offset);
pTlv->type = htons(type);

4
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -419,6 +419,7 @@ SNode* nodesMakeNode(ENodeType type) {
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
case QUERY_NODE_SHOW_TAGS_STMT:
case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
return makeNode(type, sizeof(SShowStmt));
case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
return makeNode(type, sizeof(SShowDnodeVariablesStmt));
@ -921,7 +922,8 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
case QUERY_NODE_SHOW_TAGS_STMT: {
case QUERY_NODE_SHOW_TAGS_STMT:
case QUERY_NODE_SHOW_TABLE_TAGS_STMT: {
SShowStmt* pStmt = (SShowStmt*)pNode;
nodesDestroyNode(pStmt->pDbName);
nodesDestroyNode(pStmt->pTbName);

2
source/libs/parser/inc/sql.y
View File

@ -420,6 +420,7 @@ 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 TABLE TAGS FROM table_name_cond(A) from_db_opt(B). { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TABLE_TAGS_STMT, B, A, OP_TYPE_EQUAL); }
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)); }
@ -708,6 +709,7 @@ pseudo_column(A) ::= WSTART(B).
pseudo_column(A) ::= WEND(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= WDURATION(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= IROWTS(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= QTAGS(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
function_expression(A) ::= function_name(B) NK_LP expression_list(C) NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, C)); }
function_expression(A) ::= star_func(B) NK_LP star_func_para_list(C) NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, C)); }

2
source/libs/parser/src/parAstCreater.c
View File

@ -1310,7 +1310,7 @@ SNode* createUseDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName) {
static bool needDbShowStmt(ENodeType type) {
return QUERY_NODE_SHOW_TABLES_STMT == type || QUERY_NODE_SHOW_STABLES_STMT == type ||
QUERY_NODE_SHOW_VGROUPS_STMT == type || QUERY_NODE_SHOW_INDEXES_STMT == type ||
QUERY_NODE_SHOW_TAGS_STMT == type;
QUERY_NODE_SHOW_TAGS_STMT == type || QUERY_NODE_SHOW_TABLE_TAGS_STMT == type;
}
SNode* createShowStmt(SAstCreateContext* pCxt, ENodeType type) {

7
source/libs/parser/src/parAstParser.c
View File

@ -414,6 +414,11 @@ static int32_t collectMetaKeyFromShowTags(SCollectMetaKeyCxt* pCxt, SShowStmt* p
return code;
}
static int32_t collectMetaKeyFromShowStableTags(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return collectMetaKeyFromRealTableImpl(pCxt, ((SValueNode*)pStmt->pDbName)->literal,
((SValueNode*)pStmt->pTbName)->literal, AUTH_TYPE_READ);
}
static int32_t collectMetaKeyFromShowUsers(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USERS,
pCxt->pMetaCache);
@ -595,6 +600,8 @@ static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
return collectMetaKeyFromShowTables(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_TAGS_STMT:
return collectMetaKeyFromShowTags(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
return collectMetaKeyFromShowStableTags(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_USERS_STMT:
return collectMetaKeyFromShowUsers(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_LICENCES_STMT:

1
source/libs/parser/src/parTokenizer.c
View File

@ -257,6 +257,7 @@ static SKeyword keywordTable[] = {
{"_QEND", TK_QEND},
{"_QSTART", TK_QSTART},
{"_ROWTS", TK_ROWTS},
{"_TAGS", TK_QTAGS},
{"_WDURATION", TK_WDURATION},
{"_WEND", TK_WEND},
{"_WSTART", TK_WSTART},

136
source/libs/parser/src/parTranslater.c
View File

@ -1215,6 +1215,9 @@ static bool isMultiResFunc(SNode* pNode) {
if (QUERY_NODE_FUNCTION != nodeType(pNode) || !fmIsMultiResFunc(((SFunctionNode*)pNode)->funcId)) {
return false;
}
if (FUNCTION_TYPE_TAGS == ((SFunctionNode*)pNode)->funcType) {
return true;
}
SNodeList* pParameterList = ((SFunctionNode*)pNode)->pParameterList;
if (LIST_LENGTH(pParameterList) > 1) {
return true;
@ -1556,7 +1559,7 @@ static int32_t translateMultiResFunc(STranslateContext* pCxt, SFunctionNode* pFu
"%s(*) is only supported in SELECTed list", pFunc->functionName);
}
}
if (tsKeepColumnName) {
if (tsKeepColumnName && 1 == LIST_LENGTH(pFunc->pParameterList)) {
strcpy(pFunc->node.userAlias, ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->userAlias);
}
return TSDB_CODE_SUCCESS;
@ -1842,17 +1845,46 @@ static EDealRes translateWhenThen(STranslateContext* pCxt, SWhenThenNode* pWhenT
return DEAL_RES_CONTINUE;
}
static bool isCondition(const SNode* pNode) {
if (QUERY_NODE_OPERATOR == nodeType(pNode)) {
return nodesIsComparisonOp((const SOperatorNode*)pNode);
}
return (QUERY_NODE_LOGIC_CONDITION == nodeType(pNode));
}
static int32_t rewriteIsTrue(SNode* pSrc, SNode** pIsTrue) {
SOperatorNode* pOp = (SOperatorNode*)nodesMakeNode(QUERY_NODE_OPERATOR);
if (NULL == pOp) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pOp->opType = OP_TYPE_IS_TRUE;
pOp->pLeft = pSrc;
pOp->node.resType.type = TSDB_DATA_TYPE_BOOL;
pOp->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_BOOL].bytes;
*pIsTrue = (SNode*)pOp;
return TSDB_CODE_SUCCESS;
}
static EDealRes translateCaseWhen(STranslateContext* pCxt, SCaseWhenNode* pCaseWhen) {
bool first = true;
SNode* pNode = NULL;
FOREACH(pNode, pCaseWhen->pWhenThenList) {
SWhenThenNode* pWhenThen = (SWhenThenNode*)pNode;
if (NULL == pCaseWhen->pCase && !isCondition(pWhenThen->pWhen)) {
SNode* pIsTrue = NULL;
pCxt->errCode = rewriteIsTrue(pWhenThen->pWhen, &pIsTrue);
if (TSDB_CODE_SUCCESS != pCxt->errCode) {
return DEAL_RES_ERROR;
}
pWhenThen->pWhen = pIsTrue;
}
if (first) {
pCaseWhen->node.resType = ((SExprNode*)pNode)->resType;
} else if (!dataTypeEqual(&pCaseWhen->node.resType, &((SExprNode*)pNode)->resType)) {
SWhenThenNode* pWhenThen = (SWhenThenNode*)pNode;
SNode* pCastFunc = NULL;
if (TSDB_CODE_SUCCESS != createCastFunc(pCxt, pWhenThen->pThen, pCaseWhen->node.resType, &pCastFunc)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "CASE WHEN data type mismatch");
SNode* pCastFunc = NULL;
pCxt->errCode = createCastFunc(pCxt, pWhenThen->pThen, pCaseWhen->node.resType, &pCastFunc);
if (TSDB_CODE_SUCCESS != pCxt->errCode) {
return DEAL_RES_ERROR;
}
pWhenThen->pThen = pCastFunc;
pWhenThen->node.resType = pCaseWhen->node.resType;
@ -1860,8 +1892,9 @@ static EDealRes translateCaseWhen(STranslateContext* pCxt, SCaseWhenNode* pCaseW
}
if (NULL != pCaseWhen->pElse && !dataTypeEqual(&pCaseWhen->node.resType, &((SExprNode*)pCaseWhen->pElse)->resType)) {
SNode* pCastFunc = NULL;
if (TSDB_CODE_SUCCESS != createCastFunc(pCxt, pCaseWhen->pElse, pCaseWhen->node.resType, &pCastFunc)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, "CASE WHEN data type mismatch");
pCxt->errCode = createCastFunc(pCxt, pCaseWhen->pElse, pCaseWhen->node.resType, &pCastFunc);
if (TSDB_CODE_SUCCESS != pCxt->errCode) {
return DEAL_RES_ERROR;
}
pCaseWhen->pElse = pCastFunc;
((SExprNode*)pCaseWhen->pElse)->resType = pCaseWhen->node.resType;
@ -2585,6 +2618,34 @@ static int32_t createMultiResFuncsFromStar(STranslateContext* pCxt, SFunctionNod
return code;
}
static int32_t createTags(STranslateContext* pCxt, SNodeList** pOutput) {
if (QUERY_NODE_REAL_TABLE != nodeType(((SSelectStmt*)pCxt->pCurrStmt)->pFromTable)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TAGS_PC,
"The _TAGS pseudo column can only be used for subtable and supertable queries");
}
SRealTableNode* pTable = (SRealTableNode*)(((SSelectStmt*)pCxt->pCurrStmt)->pFromTable);
const STableMeta* pMeta = pTable->pMeta;
if (TSDB_SUPER_TABLE != pMeta->tableType && TSDB_CHILD_TABLE != pMeta->tableType) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TAGS_PC,
"The _TAGS pseudo column can only be used for subtable and supertable queries");
}
SSchema* pTagsSchema = getTableTagSchema(pMeta);
for (int32_t i = 0; i < pMeta->tableInfo.numOfTags; ++i) {
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_OUT_OF_MEMORY);
}
setColumnInfoBySchema(pTable, pTagsSchema + i, 1, pCol);
if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(pOutput, (SNode*)pCol)) {
NODES_DESTORY_LIST(*pOutput);
return TSDB_CODE_OUT_OF_MEMORY;
}
}
return TSDB_CODE_SUCCESS;
}
static int32_t translateStar(STranslateContext* pCxt, SSelectStmt* pSelect) {
SNode* pNode = NULL;
WHERE_EACH(pNode, pSelect->pProjectionList) {
@ -2598,10 +2659,14 @@ static int32_t translateStar(STranslateContext* pCxt, SSelectStmt* pSelect) {
continue;
}
} else if (isMultiResFunc(pNode)) {
SNodeList* pFuncs = NULL;
code = createMultiResFuncsFromStar(pCxt, (SFunctionNode*)pNode, &pFuncs);
SNodeList* pNodeList = NULL;
if (FUNCTION_TYPE_TAGS == ((SFunctionNode*)pNode)->funcType) {
code = createTags(pCxt, &pNodeList);
} else {
code = createMultiResFuncsFromStar(pCxt, (SFunctionNode*)pNode, &pNodeList);
}
if (TSDB_CODE_SUCCESS == code) {
INSERT_LIST(pSelect->pProjectionList, pFuncs);
INSERT_LIST(pSelect->pProjectionList, pNodeList);
ERASE_NODE(pSelect->pProjectionList);
continue;
}
@ -4579,6 +4644,8 @@ static SNode* createTbnameFunction() {
return NULL;
}
strcpy(pFunc->functionName, "tbname");
strcpy(pFunc->node.aliasName, "tbname");
strcpy(pFunc->node.userAlias, "tbname");
return (SNode*)pFunc;
}
@ -6007,7 +6074,7 @@ static SNode* createProjectCol(const char* pProjCol) {
static SNodeList* createProjectCols(int32_t ncols, const char* const pCols[]) {
SNodeList* pProjections = NULL;
if (ncols <= 0) {
if (0 == ncols) {
nodesListMakeStrictAppend(&pProjections, createStarCol());
return pProjections;
}
@ -6039,10 +6106,12 @@ static int32_t createSimpleSelectStmt(const char* pDb, const char* pTable, int32
strcpy(pRealTable->table.tableAlias, pTable);
pSelect->pFromTable = (SNode*)pRealTable;
pSelect->pProjectionList = createProjectCols(numOfProjs, pProjCol);
if (NULL == pSelect->pProjectionList) {
nodesDestroyNode((SNode*)pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
if (numOfProjs >= 0) {
pSelect->pProjectionList = createProjectCols(numOfProjs, pProjCol);
if (NULL == pSelect->pProjectionList) {
nodesDestroyNode((SNode*)pSelect);
return TSDB_CODE_OUT_OF_MEMORY;
}
}
*pStmt = pSelect;
@ -6145,6 +6214,40 @@ static int32_t rewriteShow(STranslateContext* pCxt, SQuery* pQuery) {
pQuery->showRewrite = true;
nodesDestroyNode(pQuery->pRoot);
pQuery->pRoot = (SNode*)pStmt;
} else {
nodesDestroyNode((SNode*)pStmt);
}
return code;
}
static SNode* createTagsFunction() {
SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
if (NULL == pFunc) {
return NULL;
}
strcpy(pFunc->functionName, "_tags");
return (SNode*)pFunc;
}
static int32_t rewriteShowStableTags(STranslateContext* pCxt, SQuery* pQuery) {
const char* cols[] = {"tbname", "_tags"};
SShowStmt* pShow = (SShowStmt*)pQuery->pRoot;
SSelectStmt* pSelect = NULL;
int32_t code = createSimpleSelectStmt(((SValueNode*)pShow->pDbName)->literal, ((SValueNode*)pShow->pTbName)->literal,
-1, NULL, &pSelect);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeStrictAppend(&pSelect->pProjectionList, createTbnameFunction());
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeStrictAppend(&pSelect->pProjectionList, createTagsFunction());
}
if (TSDB_CODE_SUCCESS == code) {
pSelect->isDistinct = true;
pQuery->showRewrite = true;
nodesDestroyNode(pQuery->pRoot);
pQuery->pRoot = (SNode*)pSelect;
} else {
nodesDestroyNode((SNode*)pSelect);
}
return code;
}
@ -7282,6 +7385,9 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
case QUERY_NODE_SHOW_TAGS_STMT:
code = rewriteShow(pCxt, pQuery);
break;
case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
code = rewriteShowStableTags(pCxt, pQuery);
break;
case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
code = rewriteShowDnodeVariables(pCxt, pQuery);
break;

5639
source/libs/parser/src/sql.c
View File

File diff suppressed because it is too large Load Diff

2
source/libs/planner/test/planBasicTest.cpp
View File

@ -180,6 +180,8 @@ TEST_F(PlanBasicTest, pseudoColumn) {
run("SELECT _QSTART, _QEND, _QDURATION, _WSTART, _WEND, _WDURATION, COUNT(*) FROM t1 "
"WHERE ts BETWEEN '2017-7-14 18:00:00' AND '2017-7-14 19:00:00' INTERVAL(10S)");
run("SELECT _TAGS, * FROM st1s1");
}
TEST_F(PlanBasicTest, indefiniteRowsFunc) {

2
source/libs/planner/test/planOtherTest.cpp
View File

@ -86,6 +86,8 @@ TEST_F(PlanOtherTest, show) {
run("SHOW DNODE 1 VARIABLES");
run("SHOW TAGS FROM st1s1");
run("SHOW TABLE TAGS FROM st1");
}
TEST_F(PlanOtherTest, delete) {

25
source/libs/sync/src/syncMain.c
View File

@ -396,29 +396,6 @@ bool syncIsReady(int64_t rid) {
return b;
}
bool syncIsReadyForRead(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) {
return false;
}
ASSERT(rid == pSyncNode->rid);
// TODO: last not noop?
SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode);
bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && (pSyncNode->commitIndex >= lastIndex - SYNC_MAX_READ_RANGE);
taosReleaseRef(tsNodeRefId, pSyncNode->rid);
// if false, set error code
if (false == b) {
if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
terrno = TSDB_CODE_SYN_NOT_LEADER;
} else {
terrno = TSDB_CODE_APP_NOT_READY;
}
}
return b;
}
bool syncIsRestoreFinish(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) {
@ -2965,7 +2942,7 @@ static int32_t syncNodeConfigChangeFinish(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyn
cbMeta.newCfgTerm = pFinish->newCfgTerm;
cbMeta.newCfgSeqNum = pFinish->newCfgSeqNum;
ths->pFsm->FpReConfigCb(ths->pFsm, pRpcMsg, cbMeta);
ths->pFsm->FpReConfigCb(ths->pFsm, pRpcMsg, &cbMeta);
}
// clear changing

4
source/libs/sync/test/syncConfigChangeSnapshotTest.cpp
View File

@ -149,9 +149,9 @@ int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_
void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta) {
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta *cbMeta) {
sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%" PRId64 ", code:%d, currentTerm:%" PRIu64 ", term:%" PRIu64,
cbMeta.flag, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
cbMeta->flag, cbMeta->index, cbMeta->code, cbMeta->currentTerm, cbMeta->term);
}
SSyncFSM* createFsm() {

4
source/libs/sync/test/syncConfigChangeTest.cpp
View File

@ -80,9 +80,9 @@ int32_t GetSnapshotCb(struct SSyncFSM* pFsm, SSnapshot* pSnapshot) {
void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta) {
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta *cbMeta) {
sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%" PRId64 ", code:%d, currentTerm:%" PRIu64 ", term:%" PRIu64,
cbMeta.flag, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
cbMeta->flag, cbMeta->index, cbMeta->code, cbMeta->currentTerm, cbMeta->term);
}
SSyncFSM* createFsm() {

4
source/libs/sync/test/syncTestTool.cpp
View File

@ -153,11 +153,11 @@ int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_
void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb== pFsm:%p", pFsm); }
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta) {
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta* cbMeta) {
char* s = syncCfg2Str(&(cbMeta.newCfg));
sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%" PRId64 ", code:%d, currentTerm:%" PRIu64 ", term:%" PRIu64
", newCfg:%s",
cbMeta.flag, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term, s);
cbMeta->flag, cbMeta->index, cbMeta->code, cbMeta->currentTerm, cbMeta->term, s);
taosMemoryFree(s);
}

4
tests/docs-examples-test/node.sh
View File

@ -23,7 +23,7 @@ node query_example.js
node async_query_example.js
node subscribe_demo.js
# node subscribe_demo.js
taos -s "drop topic if exists topic_name_example"
taos -s "drop database if exists power"
@ -39,4 +39,4 @@ taos -s "drop database if exists test"
node opentsdb_telnet_example.js
taos -s "drop database if exists test"
node opentsdb_json_example.js
node opentsdb_json_example.js

2
tests/script/jenkins/basic.txt
View File

@ -303,7 +303,7 @@
./test.sh -f tsim/insert/backquote.sim -m
# unsupport ./test.sh -f tsim/parser/fourArithmetic-basic.sim -m
./test.sh -f tsim/query/interval-offset.sim -m
./test.sh -f tsim/tmq/basic3.sim -m
# unsupport ./test.sh -f tsim/tmq/basic3.sim -m
./test.sh -f tsim/stable/vnode3.sim -m
./test.sh -f tsim/qnode/basic1.sim -m
# unsupport ./test.sh -f tsim/mnode/basic1.sim -m