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

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

This commit is contained in:
Hongze Cheng 2022-09-30 13:50:01 +08:00
parent 145abcb17f 1f57a5a1cc
commit 7754471aaa
55 changed files with 1000 additions and 840 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docs/en/07-develop/03-insert-data/03-opentsdb-telnet.mdx
View File

@ -64,10 +64,10 @@ taos> use test;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
meters.current | 2022-03-30 17:04:10.877 | 2 | 2 | 2 | meters.current |
meters.voltage | 2022-03-30 17:04:10.882 | 2 | 2 | 2 | meters.voltage |
Query OK, 2 row(s) in set (0.002544s) Query OK, 2 row(s) in set (0.002544s)
taos> select tbname, * from `meters.current`; taos> select tbname, * from `meters.current`;

8
docs/en/07-develop/03-insert-data/04-opentsdb-json.mdx
View File

@ -81,10 +81,10 @@ taos> use test;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
meters.current | 2022-03-29 16:05:25.193 | 2 | 2 | 1 | meters.current |
meters.voltage | 2022-03-29 16:05:25.200 | 2 | 2 | 1 | meters.voltage |
Query OK, 2 row(s) in set (0.001954s) Query OK, 2 row(s) in set (0.001954s)
taos> select * from `meters.current`; taos> select * from `meters.current`;

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

@ -4,11 +4,11 @@ Execute TDengine CLI program `taos` directly from the Linux shell to connect to
$ taos $ taos
taos> show databases; taos> show databases;
name | create_time | vgroups | ntables | replica | strict | duration | keep | buffer | pagesize | pages | minrows | maxrows | comp | precision | status | retention | single_stable | cachemodel | cachesize | wal_level | wal_fsync_period | wal_retention_period | wal_retention_size | wal_roll_period | wal_seg_size | name |
========================================================================================================================================================================================================================================================================================================================================================================================================================================================================= =================================
information_schema | NULL | NULL | 14 | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | ready | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | information_schema |
performance_schema | NULL | NULL | 3 | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | ready | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | performance_schema |
db | 2022-08-04 14:14:49.385 | 2 | 4 | 1 | off | 14400m | 5254560m,5254560m,5254560m | 96 | 4 | 256 | 100 | 4096 | 2 | ms | ready | NULL | false | none | 1 | 1 | 3000 | 0 | 0 | 0 | 0 | db |
Query OK, 3 rows in database (0.019154s) Query OK, 3 rows in database (0.019154s)
taos> taos>

11
docs/en/14-reference/03-connector/_verify_windows.mdx
View File

@ -2,12 +2,11 @@ Go to the `C:\TDengine` directory from `cmd` and execute TDengine CLI program `t
```text ```text
taos> show databases; taos> show databases;
name | create_time | vgroups | ntables | replica | strict | duration | keep | buffer | pagesize | pages | minrows | maxrows | comp | precision | status | retention | single_stable | cachemodel | cachesize | wal_level | wal_fsync_period | wal_retention_period | wal_retention_size | wal_roll_period | wal_seg_size | name |
========================================================================================================================================================================================================================================================================================================================================================================================================================================================================= =================================
information_schema | NULL | NULL | 14 | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | ready | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | information_schema |
performance_schema | NULL | NULL | 3 | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | ready | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | performance_schema |
test | 2022-08-04 16:46:40.506 | 2 | 0 | 1 | off | 14400m | 5256000m,5256000m,5256000m | 96 | 4 | 256 | test |
100 | 4096 | 2 | ms | ready | NULL | false | none | 1 | 1 | 3000 | 0 | 0 | 0 | 0 |
Query OK, 3 rows in database (0.123000s) Query OK, 3 rows in database (0.123000s)
taos> taos>

9
docs/en/14-reference/11-docker/index.md
View File

@ -25,10 +25,11 @@ The TDengine client taos can be executed in this container to access TDengine us
$ docker exec -it tdengine taos $ docker exec -it tdengine taos
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
log | 2022-01-17 13:57:22.270 | 10 | 1 | 1 | 1 | 10 | 30 | 1 | 3 | 100 | 4096 | 1 | 3000 | 2 | 0 | us | 0 | ready | information_schema |
Query OK, 1 row(s) in set (0.002843s) performance_schema |
Query OK, 2 row(s) in set (0.002843s)
``` ```
The TDengine server running in the container uses the container's hostname to establish a connection. Using TDengine CLI or various connectors (such as JDBC-JNI) to access the TDengine inside the container from outside the container is more complicated. So the above is the simplest way to access the TDengine service in the container and is suitable for some simple scenarios. Please refer to the next section if you want to access the TDengine service in the container from outside the container using TDengine CLI or various connectors for complex scenarios. The TDengine server running in the container uses the container's hostname to establish a connection. Using TDengine CLI or various connectors (such as JDBC-JNI) to access the TDengine inside the container from outside the container is more complicated. So the above is the simplest way to access the TDengine service in the container and is suitable for some simple scenarios. Please refer to the next section if you want to access the TDengine service in the container from outside the container using TDengine CLI or various connectors for complex scenarios.

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

@ -51,5 +51,6 @@ port: 8125
Start StatsD after adding the following (assuming the config file is modified to config.js) Start StatsD after adding the following (assuming the config file is modified to config.js)
``` ```
npm install
node stats.js config.js & node stats.js config.js &
``` ```

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

@ -30,21 +30,20 @@ After restarting Prometheus, you can refer to the following example to verify th
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
test | 2022-04-12 08:07:58.756 | 1 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready | information_schema |
log | 2022-04-20 07:19:50.260 | 2 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready | performance_schema |
prometheus_data | 2022-04-20 07:21:09.202 | 158 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | prometheus_data |
db | 2022-04-15 06:37:08.512 | 1 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready | Query OK, 3 row(s) in set (0.000585s)
Query OK, 4 row(s) in set (0.000585s)
taos> use prometheus_data; taos> use prometheus_data;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
metrics | 2022-04-20 07:21:09.209 | 2 | 1 | 1389 | metrics |
Query OK, 1 row(s) in set (0.000487s) Query OK, 1 row(s) in set (0.000487s)
taos> select * from metrics limit 10; taos> select * from metrics limit 10;

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

@ -43,16 +43,16 @@ taos> use telegraf;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
swap | 2022-04-20 08:47:53.532 | 7 | 1 | 1 | swap |
cpu | 2022-04-20 08:48:03.488 | 11 | 2 | 5 | cpu |
system | 2022-04-20 08:47:53.512 | 8 | 1 | 1 | system |
diskio | 2022-04-20 08:47:53.550 | 12 | 2 | 15 | diskio |
kernel | 2022-04-20 08:47:53.503 | 6 | 1 | 1 | kernel |
mem | 2022-04-20 08:47:53.521 | 35 | 1 | 1 | mem |
processes | 2022-04-20 08:47:53.555 | 12 | 1 | 1 | processes |
disk | 2022-04-20 08:47:53.541 | 8 | 5 | 2 | disk |
Query OK, 8 row(s) in set (0.000521s) Query OK, 8 row(s) in set (0.000521s)
taos> select * from telegraf.system limit 10; taos> select * from telegraf.system limit 10;

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

@ -32,28 +32,29 @@ Use the TDengine CLI to verify that collectd's data is written to TDengine and c
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
collectd | 2022-04-20 09:27:45.460 | 95 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | information_schema |
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 | performance_schema |
Query OK, 2 row(s) in set (0.003266s) collectd |
Query OK, 3 row(s) in set (0.003266s)
taos> use collectd; taos> use collectd;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
load_1 | 2022-04-20 09:27:45.492 | 2 | 2 | 1 | load_1 |
memory_value | 2022-04-20 09:27:45.463 | 2 | 3 | 6 | memory_value |
df_value | 2022-04-20 09:27:45.463 | 2 | 4 | 25 | df_value |
load_2 | 2022-04-20 09:27:45.501 | 2 | 2 | 1 | load_2 |
load_0 | 2022-04-20 09:27:45.485 | 2 | 2 | 1 | load_0 |
interface_1 | 2022-04-20 09:27:45.488 | 2 | 3 | 12 | interface_1 |
irq_value | 2022-04-20 09:27:45.476 | 2 | 3 | 31 | irq_value |
interface_0 | 2022-04-20 09:27:45.480 | 2 | 3 | 12 | interface_0 |
entropy_value | 2022-04-20 09:27:45.473 | 2 | 2 | 1 | entropy_value |
swap_value | 2022-04-20 09:27:45.477 | 2 | 3 | 5 | swap_value |
Query OK, 10 row(s) in set (0.002236s) Query OK, 10 row(s) in set (0.002236s)
taos> select * from collectd.memory_value limit 10; taos> select * from collectd.memory_value limit 10;

24
docs/en/20-third-party/06-statsd.md vendored
View File

@ -26,7 +26,7 @@ Start StatsD:
``` ```
$ node stats.js config.js & $ node stats.js config.js &
[1] 8546 [1] 8546
$ 20 Apr 09:54:41 - [8546] reading config file: exampleConfig.js $ 20 Apr 09:54:41 - [8546] reading config file: config.js
20 Apr 09:54:41 - server is up INFO 20 Apr 09:54:41 - server is up INFO
``` ```
@ -40,19 +40,20 @@ Use the TDengine CLI to verify that StatsD data is written to TDengine and can r
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
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 | information_schema |
statsd | 2022-04-20 09:54:51.220 | 1 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | performance_schema |
Query OK, 2 row(s) in set (0.003142s) statsd |
Query OK, 3 row(s) in set (0.003142s)
taos> use statsd; taos> use statsd;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
foo | 2022-04-20 09:54:51.234 | 2 | 1 | 1 | foo |
Query OK, 1 row(s) in set (0.002161s) Query OK, 1 row(s) in set (0.002161s)
taos> select * from foo; taos> select * from foo;
@ -63,3 +64,8 @@ Query OK, 1 row(s) in set (0.004179s)
taos> taos>
``` ```
:::note
- TDengine will automatically create unique IDs for sub-table names by the rule.
:::

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

@ -47,30 +47,30 @@ taos> use icinga2;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
icinga.service.users.state_... | 2022-04-20 12:11:39.726 | 2 | 1 | 1 | icinga.service.users.state_... |
icinga.service.users.acknow... | 2022-04-20 12:11:39.756 | 2 | 1 | 1 | icinga.service.users.acknow... |
icinga.service.procs.downti... | 2022-04-20 12:11:44.541 | 2 | 1 | 1 | icinga.service.procs.downti... |
icinga.service.users.users | 2022-04-20 12:11:39.770 | 2 | 1 | 1 | icinga.service.users.users |
icinga.service.procs.procs_min | 2022-04-20 12:11:44.599 | 2 | 1 | 1 | icinga.service.procs.procs_min |
icinga.service.users.users_min | 2022-04-20 12:11:39.809 | 2 | 1 | 1 | icinga.service.users.users_min |
icinga.check.max_check_atte... | 2022-04-20 12:11:39.847 | 2 | 3 | 2 | icinga.check.max_check_atte... |
icinga.service.procs.state_... | 2022-04-20 12:11:44.522 | 2 | 1 | 1 | icinga.service.procs.state_... |
icinga.service.procs.procs_... | 2022-04-20 12:11:44.576 | 2 | 1 | 1 | icinga.service.procs.procs_... |
icinga.service.users.users_... | 2022-04-20 12:11:39.796 | 2 | 1 | 1 | icinga.service.users.users_... |
icinga.check.latency | 2022-04-20 12:11:39.869 | 2 | 3 | 2 | icinga.check.latency |
icinga.service.procs.procs_... | 2022-04-20 12:11:44.588 | 2 | 1 | 1 | icinga.service.procs.procs_... |
icinga.service.users.downti... | 2022-04-20 12:11:39.746 | 2 | 1 | 1 | icinga.service.users.downti... |
icinga.service.users.users_... | 2022-04-20 12:11:39.783 | 2 | 1 | 1 | icinga.service.users.users_... |
icinga.service.users.reachable | 2022-04-20 12:11:39.736 | 2 | 1 | 1 | icinga.service.users.reachable |
icinga.service.procs.procs | 2022-04-20 12:11:44.565 | 2 | 1 | 1 | icinga.service.procs.procs |
icinga.service.procs.acknow... | 2022-04-20 12:11:44.554 | 2 | 1 | 1 | icinga.service.procs.acknow... |
icinga.service.procs.state | 2022-04-20 12:11:44.509 | 2 | 1 | 1 | icinga.service.procs.state |
icinga.service.procs.reachable | 2022-04-20 12:11:44.532 | 2 | 1 | 1 | icinga.service.procs.reachable |
icinga.check.current_attempt | 2022-04-20 12:11:39.825 | 2 | 3 | 2 | icinga.check.current_attempt |
icinga.check.execution_time | 2022-04-20 12:11:39.898 | 2 | 3 | 2 | icinga.check.execution_time |
icinga.service.users.state | 2022-04-20 12:11:39.704 | 2 | 1 | 1 | icinga.service.users.state |
Query OK, 22 row(s) in set (0.002317s) Query OK, 22 row(s) in set (0.002317s)
``` ```

56
docs/en/20-third-party/08-tcollector.md vendored
View File

@ -33,35 +33,41 @@ Wait for a few seconds and then use the TDengine CLI to query whether the corres
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
tcollector | 2022-04-20 12:44:49.604 | 88 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | information_schema |
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 | performance_schema |
Query OK, 2 row(s) in set (0.002679s) tcollector |
Query OK, 3 rows in database (0.001647s)
taos> use tcollector; taos> use tcollector;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
proc.meminfo.hugepages_rsvd | 2022-04-20 12:44:53.945 | 2 | 1 | 1 | proc.meminfo.hugepages_rsvd |
proc.meminfo.directmap1g | 2022-04-20 12:44:54.110 | 2 | 1 | 1 | proc.meminfo.directmap1g |
proc.meminfo.vmallocchunk | 2022-04-20 12:44:53.724 | 2 | 1 | 1 | proc.meminfo.vmallocchunk |
proc.meminfo.hugepagesize | 2022-04-20 12:44:54.004 | 2 | 1 | 1 | proc.meminfo.hugepagesize |
tcollector.reader.lines_dro... | 2022-04-20 12:44:49.675 | 2 | 1 | 1 | tcollector.reader.lines_dro... |
proc.meminfo.sunreclaim | 2022-04-20 12:44:53.437 | 2 | 1 | 1 | proc.meminfo.sunreclaim |
proc.stat.ctxt | 2022-04-20 12:44:55.363 | 2 | 1 | 1 | proc.stat.ctxt |
proc.meminfo.swaptotal | 2022-04-20 12:44:53.158 | 2 | 1 | 1 | proc.meminfo.swaptotal |
proc.uptime.total | 2022-04-20 12:44:52.813 | 2 | 1 | 1 | proc.uptime.total |
tcollector.collector.lines_... | 2022-04-20 12:44:49.895 | 2 | 2 | 51 | tcollector.collector.lines_... |
proc.meminfo.vmallocused | 2022-04-20 12:44:53.704 | 2 | 1 | 1 | proc.meminfo.vmallocused |
proc.meminfo.memavailable | 2022-04-20 12:44:52.939 | 2 | 1 | 1 | proc.meminfo.memavailable |
sys.numa.foreign_allocs | 2022-04-20 12:44:57.929 | 2 | 2 | 1 | sys.numa.foreign_allocs |
proc.meminfo.committed_as | 2022-04-20 12:44:53.639 | 2 | 1 | 1 | proc.meminfo.committed_as |
proc.vmstat.pswpin | 2022-04-20 12:44:54.177 | 2 | 1 | 1 | proc.vmstat.pswpin |
proc.meminfo.cmafree | 2022-04-20 12:44:53.865 | 2 | 1 | 1 | proc.meminfo.cmafree |
proc.meminfo.mapped | 2022-04-20 12:44:53.349 | 2 | 1 | 1 | proc.meminfo.mapped |
proc.vmstat.pgmajfault | 2022-04-20 12:44:54.251 | 2 | 1 | 1 | proc.vmstat.pgmajfault |
... ...
``` ```
:::note
- TDengine will automatically create unique IDs for sub-table names by the rule.
:::

8
docs/zh/07-develop/03-insert-data/03-opentsdb-telnet.mdx
View File

@ -66,10 +66,10 @@ taos> use test;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
meters.current | 2022-03-30 17:04:10.877 | 2 | 2 | 2 | meters.current |
meters.voltage | 2022-03-30 17:04:10.882 | 2 | 2 | 2 | meters.voltage |
Query OK, 2 row(s) in set (0.002544s) Query OK, 2 row(s) in set (0.002544s)
taos> select tbname, * from `meters.current`; taos> select tbname, * from `meters.current`;

8
docs/zh/07-develop/03-insert-data/04-opentsdb-json.mdx
View File

@ -82,10 +82,10 @@ taos> use test;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
meters.current | 2022-03-29 16:05:25.193 | 2 | 2 | 1 | meters.current |
meters.voltage | 2022-03-29 16:05:25.200 | 2 | 2 | 1 | meters.voltage |
Query OK, 2 row(s) in set (0.001954s) Query OK, 2 row(s) in set (0.001954s)
taos> select * from `meters.current`; taos> select * from `meters.current`;

10
docs/zh/08-connector/_verify_linux.mdx
View File

@ -4,11 +4,11 @@
$ taos $ taos
taos> show databases; taos> show databases;
name | create_time | vgroups | ntables | replica | strict | duration | keep | buffer | pagesize | pages | minrows | maxrows | comp | precision | status | retention | single_stable | cachemodel | cachesize | wal_level | wal_fsync_period | wal_retention_period | wal_retention_size | wal_roll_period | wal_seg_size | name |
========================================================================================================================================================================================================================================================================================================================================================================================================================================================================= =================================
information_schema | NULL | NULL | 14 | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | ready | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | information_schema |
performance_schema | NULL | NULL | 3 | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | ready | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | NULL | performance_schema |
db | 2022-08-04 14:14:49.385 | 2 | 4 | 1 | off | 14400m | 5254560m,5254560m,5254560m | 96 | 4 | 256 | 100 | 4096 | 2 | ms | ready | NULL | false | none | 1 | 1 | 3000 | 0 | 0 | 0 | 0 | db |
Query OK, 3 rows in database (0.019154s) Query OK, 3 rows in database (0.019154s)
taos> taos>

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

@ -51,5 +51,6 @@ port: 8125
增加如下内容后启动 StatsD（假设配置文件修改为 config.js）。 增加如下内容后启动 StatsD（假设配置文件修改为 config.js）。
``` ```
npm install
node stats.js config.js & node stats.js config.js &
``` ```

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

@ -29,21 +29,20 @@ Prometheus 提供了 `remote_write` 和 `remote_read` 接口来利用其它数
### 使用 TDengine CLI 查询写入数据 ### 使用 TDengine CLI 查询写入数据
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
test | 2022-04-12 08:07:58.756 | 1 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready | information_schema |
log | 2022-04-20 07:19:50.260 | 2 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready | performance_schema |
prometheus_data | 2022-04-20 07:21:09.202 | 158 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | prometheus_data |
db | 2022-04-15 06:37:08.512 | 1 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ms | 0 | ready | Query OK, 3 row(s) in set (0.000585s)
Query OK, 4 row(s) in set (0.000585s)
taos> use prometheus_data; taos> use prometheus_data;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
metrics | 2022-04-20 07:21:09.209 | 2 | 1 | 1389 | metrics |
Query OK, 1 row(s) in set (0.000487s) Query OK, 1 row(s) in set (0.000487s)
taos> select * from metrics limit 10; taos> select * from metrics limit 10;

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

@ -44,16 +44,16 @@ taos> use telegraf;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
swap | 2022-04-20 08:47:53.532 | 7 | 1 | 1 | swap |
cpu | 2022-04-20 08:48:03.488 | 11 | 2 | 5 | cpu |
system | 2022-04-20 08:47:53.512 | 8 | 1 | 1 | system |
diskio | 2022-04-20 08:47:53.550 | 12 | 2 | 15 | diskio |
kernel | 2022-04-20 08:47:53.503 | 6 | 1 | 1 | kernel |
mem | 2022-04-20 08:47:53.521 | 35 | 1 | 1 | mem |
processes | 2022-04-20 08:47:53.555 | 12 | 1 | 1 | processes |
disk | 2022-04-20 08:47:53.541 | 8 | 5 | 2 | disk |
Query OK, 8 row(s) in set (0.000521s) Query OK, 8 row(s) in set (0.000521s)
taos> select * from telegraf.system limit 10; taos> select * from telegraf.system limit 10;

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

@ -32,28 +32,29 @@ sudo systemctl restart collectd
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
collectd | 2022-04-20 09:27:45.460 | 95 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | information_schema |
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 | performance_schema |
Query OK, 2 row(s) in set (0.003266s) collectd |
Query OK, 3 row(s) in set (0.003266s)
taos> use collectd; taos> use collectd;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
load_1 | 2022-04-20 09:27:45.492 | 2 | 2 | 1 | load_1 |
memory_value | 2022-04-20 09:27:45.463 | 2 | 3 | 6 | memory_value |
df_value | 2022-04-20 09:27:45.463 | 2 | 4 | 25 | df_value |
load_2 | 2022-04-20 09:27:45.501 | 2 | 2 | 1 | load_2 |
load_0 | 2022-04-20 09:27:45.485 | 2 | 2 | 1 | load_0 |
interface_1 | 2022-04-20 09:27:45.488 | 2 | 3 | 12 | interface_1 |
irq_value | 2022-04-20 09:27:45.476 | 2 | 3 | 31 | irq_value |
interface_0 | 2022-04-20 09:27:45.480 | 2 | 3 | 12 | interface_0 |
entropy_value | 2022-04-20 09:27:45.473 | 2 | 2 | 1 | entropy_value |
swap_value | 2022-04-20 09:27:45.477 | 2 | 3 | 5 | swap_value |
Query OK, 10 row(s) in set (0.002236s) Query OK, 10 row(s) in set (0.002236s)
taos> select * from collectd.memory_value limit 10; taos> select * from collectd.memory_value limit 10;

24
docs/zh/20-third-party/06-statsd.md vendored
View File

@ -27,7 +27,7 @@ StatsD 是汇总和总结应用指标的一个简单的守护进程，近些年
``` ```
$ node stats.js config.js & $ node stats.js config.js &
[1] 8546 [1] 8546
$ 20 Apr 09:54:41 - [8546] reading config file: exampleConfig.js $ 20 Apr 09:54:41 - [8546] reading config file: config.js
20 Apr 09:54:41 - server is up INFO 20 Apr 09:54:41 - server is up INFO
``` ```
@ -41,19 +41,20 @@ $ echo "foo:1|c" | nc -u -w0 127.0.0.1 8125
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
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 | information_schema |
statsd | 2022-04-20 09:54:51.220 | 1 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | performance_schema |
Query OK, 2 row(s) in set (0.003142s) statsd |
Query OK, 3 row(s) in set (0.003142s)
taos> use statsd; taos> use statsd;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
foo | 2022-04-20 09:54:51.234 | 2 | 1 | 1 | foo |
Query OK, 1 row(s) in set (0.002161s) Query OK, 1 row(s) in set (0.002161s)
taos> select * from foo; taos> select * from foo;
@ -64,3 +65,8 @@ Query OK, 1 row(s) in set (0.004179s)
taos> taos>
``` ```
:::note
- TDengine will automatically create unique IDs for sub-table names by the rule.
:::

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

@ -48,30 +48,30 @@ taos> use icinga2;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
icinga.service.users.state_... | 2022-04-20 12:11:39.726 | 2 | 1 | 1 | icinga.service.users.state_... |
icinga.service.users.acknow... | 2022-04-20 12:11:39.756 | 2 | 1 | 1 | icinga.service.users.acknow... |
icinga.service.procs.downti... | 2022-04-20 12:11:44.541 | 2 | 1 | 1 | icinga.service.procs.downti... |
icinga.service.users.users | 2022-04-20 12:11:39.770 | 2 | 1 | 1 | icinga.service.users.users |
icinga.service.procs.procs_min | 2022-04-20 12:11:44.599 | 2 | 1 | 1 | icinga.service.procs.procs_min |
icinga.service.users.users_min | 2022-04-20 12:11:39.809 | 2 | 1 | 1 | icinga.service.users.users_min |
icinga.check.max_check_atte... | 2022-04-20 12:11:39.847 | 2 | 3 | 2 | icinga.check.max_check_atte... |
icinga.service.procs.state_... | 2022-04-20 12:11:44.522 | 2 | 1 | 1 | icinga.service.procs.state_... |
icinga.service.procs.procs_... | 2022-04-20 12:11:44.576 | 2 | 1 | 1 | icinga.service.procs.procs_... |
icinga.service.users.users_... | 2022-04-20 12:11:39.796 | 2 | 1 | 1 | icinga.service.users.users_... |
icinga.check.latency | 2022-04-20 12:11:39.869 | 2 | 3 | 2 | icinga.check.latency |
icinga.service.procs.procs_... | 2022-04-20 12:11:44.588 | 2 | 1 | 1 | icinga.service.procs.procs_... |
icinga.service.users.downti... | 2022-04-20 12:11:39.746 | 2 | 1 | 1 | icinga.service.users.downti... |
icinga.service.users.users_... | 2022-04-20 12:11:39.783 | 2 | 1 | 1 | icinga.service.users.users_... |
icinga.service.users.reachable | 2022-04-20 12:11:39.736 | 2 | 1 | 1 | icinga.service.users.reachable |
icinga.service.procs.procs | 2022-04-20 12:11:44.565 | 2 | 1 | 1 | icinga.service.procs.procs |
icinga.service.procs.acknow... | 2022-04-20 12:11:44.554 | 2 | 1 | 1 | icinga.service.procs.acknow... |
icinga.service.procs.state | 2022-04-20 12:11:44.509 | 2 | 1 | 1 | icinga.service.procs.state |
icinga.service.procs.reachable | 2022-04-20 12:11:44.532 | 2 | 1 | 1 | icinga.service.procs.reachable |
icinga.check.current_attempt | 2022-04-20 12:11:39.825 | 2 | 3 | 2 | icinga.check.current_attempt |
icinga.check.execution_time | 2022-04-20 12:11:39.898 | 2 | 3 | 2 | icinga.check.execution_time |
icinga.service.users.state | 2022-04-20 12:11:39.704 | 2 | 1 | 1 | icinga.service.users.state |
Query OK, 22 row(s) in set (0.002317s) Query OK, 22 row(s) in set (0.002317s)
``` ```

57
docs/zh/20-third-party/08-tcollector.md vendored
View File

@ -34,35 +34,42 @@ sudo systemctl restart taosadapter
``` ```
taos> show databases; taos> show databases;
name | created_time | ntables | vgroups | replica | quorum | days | keep | cache(MB) | blocks | minrows | maxrows | wallevel | fsync | comp | cachelast | precision | update | status | name |
==================================================================================================================================================================================================================================================================================== =================================
tcollector | 2022-04-20 12:44:49.604 | 88 | 1 | 1 | 1 | 10 | 3650 | 16 | 6 | 100 | 4096 | 1 | 3000 | 2 | 0 | ns | 2 | ready | information_schema |
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 | performance_schema |
Query OK, 2 row(s) in set (0.002679s) tcollector |
Query OK, 3 rows in database (0.001647s)
taos> use tcollector; taos> use tcollector;
Database changed. Database changed.
taos> show stables; taos> show stables;
name | created_time | columns | tags | tables | name |
============================================================================================ =================================
proc.meminfo.hugepages_rsvd | 2022-04-20 12:44:53.945 | 2 | 1 | 1 | proc.meminfo.hugepages_rsvd |
proc.meminfo.directmap1g | 2022-04-20 12:44:54.110 | 2 | 1 | 1 | proc.meminfo.directmap1g |
proc.meminfo.vmallocchunk | 2022-04-20 12:44:53.724 | 2 | 1 | 1 | proc.meminfo.vmallocchunk |
proc.meminfo.hugepagesize | 2022-04-20 12:44:54.004 | 2 | 1 | 1 | proc.meminfo.hugepagesize |
tcollector.reader.lines_dro... | 2022-04-20 12:44:49.675 | 2 | 1 | 1 | tcollector.reader.lines_dro... |
proc.meminfo.sunreclaim | 2022-04-20 12:44:53.437 | 2 | 1 | 1 | proc.meminfo.sunreclaim |
proc.stat.ctxt | 2022-04-20 12:44:55.363 | 2 | 1 | 1 | proc.stat.ctxt |
proc.meminfo.swaptotal | 2022-04-20 12:44:53.158 | 2 | 1 | 1 | proc.meminfo.swaptotal |
proc.uptime.total | 2022-04-20 12:44:52.813 | 2 | 1 | 1 | proc.uptime.total |
tcollector.collector.lines_... | 2022-04-20 12:44:49.895 | 2 | 2 | 51 | tcollector.collector.lines_... |
proc.meminfo.vmallocused | 2022-04-20 12:44:53.704 | 2 | 1 | 1 | proc.meminfo.vmallocused |
proc.meminfo.memavailable | 2022-04-20 12:44:52.939 | 2 | 1 | 1 | proc.meminfo.memavailable |
sys.numa.foreign_allocs | 2022-04-20 12:44:57.929 | 2 | 2 | 1 | sys.numa.foreign_allocs |
proc.meminfo.committed_as | 2022-04-20 12:44:53.639 | 2 | 1 | 1 | proc.meminfo.committed_as |
proc.vmstat.pswpin | 2022-04-20 12:44:54.177 | 2 | 1 | 1 | proc.vmstat.pswpin |
proc.meminfo.cmafree | 2022-04-20 12:44:53.865 | 2 | 1 | 1 | proc.meminfo.cmafree |
proc.meminfo.mapped | 2022-04-20 12:44:53.349 | 2 | 1 | 1 | proc.meminfo.mapped |
proc.vmstat.pgmajfault | 2022-04-20 12:44:54.251 | 2 | 1 | 1 | proc.vmstat.pgmajfault |
... ...
``` ```
:::note
- TDengine 默认生成的子表名是根据规则生成的唯一 ID 值。
:::

12
include/libs/index/index.h
View File

@ -206,12 +206,6 @@ void indexJsonRebuild(SIndexJson* idx, void* iter);
**/ **/
bool indexJsonIsRebuild(SIndexJson* idx); bool indexJsonIsRebuild(SIndexJson* idx);
/*
* init index env
*
*/
void indexInit();
/* index filter */ /* index filter */
typedef struct SIndexMetaArg { typedef struct SIndexMetaArg {
void* metaEx; void* metaEx;
@ -225,6 +219,12 @@ typedef enum { SFLT_NOT_INDEX, SFLT_COARSE_INDEX, SFLT_ACCURATE_INDEX } SIdxFltS
SIdxFltStatus idxGetFltStatus(SNode* pFilterNode); SIdxFltStatus idxGetFltStatus(SNode* pFilterNode);
int32_t doFilterTag(SNode* pFilterNode, SIndexMetaArg* metaArg, SArray* result, SIdxFltStatus* status); int32_t doFilterTag(SNode* pFilterNode, SIndexMetaArg* metaArg, SArray* result, SIdxFltStatus* status);
/*
* init index env
*
*/
void indexInit(int32_t threads);
/* /*
* destory index env * destory index env
* *

6
include/libs/stream/streamState.h
View File

@ -33,6 +33,7 @@ typedef struct {
TTB* pFuncStateDb; TTB* pFuncStateDb;
TTB* pFillStateDb; // todo refactor TTB* pFillStateDb; // todo refactor
TXN txn; TXN txn;
int32_t number;
} SStreamState; } SStreamState;
SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath); SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath);
@ -43,6 +44,7 @@ int32_t streamStateAbort(SStreamState* pState);
typedef struct { typedef struct {
TBC* pCur; TBC* pCur;
int64_t number;
} SStreamStateCur; } SStreamStateCur;
int32_t streamStateFuncPut(SStreamState* pState, const STupleKey* key, const void* value, int32_t vLen); int32_t streamStateFuncPut(SStreamState* pState, const STupleKey* key, const void* value, int32_t vLen);
@ -52,6 +54,8 @@ int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key);
int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen); int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen); int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateDel(SStreamState* pState, const SWinKey* key); int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
int32_t streamStateClear(SStreamState* pState);
void streamStateSetNumber(SStreamState* pState, int32_t number);
int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen); int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen); int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
@ -63,6 +67,7 @@ void streamFreeVal(void* val);
SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key); SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key); SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key);
SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key); SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey* key); SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey* key);
void streamStateFreeCur(SStreamStateCur* pCur); void streamStateFreeCur(SStreamStateCur* pCur);
@ -70,6 +75,7 @@ void streamStateFreeCur(SStreamStateCur* pCur);
int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen); int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen); int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
int32_t streamStateGetFirst(SStreamState* pState, SWinKey* key);
int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur); int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur);
int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur); int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);

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

@ -193,6 +193,8 @@ int32_t dmInitDnode(SDnode *pDnode, EDndNodeType rtype) {
goto _OVER; goto _OVER;
} }
indexInit(tsNumOfCommitThreads);
dmReportStartup("dnode-transport", "initialized"); dmReportStartup("dnode-transport", "initialized");
dDebug("dnode is created, ptr:%p", pDnode); dDebug("dnode is created, ptr:%p", pDnode);
code = 0; code = 0;

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

@ -237,6 +237,7 @@ static int32_t mndCreateDefaultCluster(SMnode *pMnode) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, NULL, "create-cluster"); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, NULL, "create-cluster");
if (pTrans == NULL) { if (pTrans == NULL) {
sdbFreeRaw(pRaw);
mError("cluster:%" PRId64 ", failed to create since %s", clusterObj.id, terrstr()); mError("cluster:%" PRId64 ", failed to create since %s", clusterObj.id, terrstr());
return -1; return -1;
} }

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

@ -1594,7 +1594,7 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
break; break;
} }
char precVstr[10] = {0}; char precVstr[10] = {0};
STR_WITH_SIZE_TO_VARSTR(precVstr, precStr, 2); STR_WITH_MAXSIZE_TO_VARSTR(precVstr, precStr, 10);
char *statusStr = "ready"; char *statusStr = "ready";
if (objStatus == SDB_STATUS_CREATING) { if (objStatus == SDB_STATUS_CREATING) {
@ -1607,7 +1607,7 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
} }
} }
char statusVstr[24] = {0}; char statusVstr[24] = {0};
STR_WITH_SIZE_TO_VARSTR(statusVstr, statusStr, strlen(statusStr)); STR_WITH_MAXSIZE_TO_VARSTR(statusVstr, statusStr, 24);
if (sysDb || !sysinfo) { if (sysDb || !sysinfo) {
for (int32_t i = 0; i < pShow->numOfColumns; ++i) { for (int32_t i = 0; i < pShow->numOfColumns; ++i) {
@ -1644,7 +1644,7 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
const char *strictStr = pDb->cfg.strict ? "on" : "off"; const char *strictStr = pDb->cfg.strict ? "on" : "off";
char strictVstr[24] = {0}; char strictVstr[24] = {0};
STR_WITH_SIZE_TO_VARSTR(strictVstr, strictStr, strlen(strictStr)); STR_WITH_MAXSIZE_TO_VARSTR(strictVstr, strictStr, 24);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, rows, (const char *)strictVstr, false); colDataAppend(pColInfo, rows, (const char *)strictVstr, false);
@ -1704,7 +1704,7 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
const char *cacheModelStr = getCacheModelStr(pDb->cfg.cacheLast); const char *cacheModelStr = getCacheModelStr(pDb->cfg.cacheLast);
char cacheModelVstr[24] = {0}; char cacheModelVstr[24] = {0};
STR_WITH_SIZE_TO_VARSTR(cacheModelVstr, cacheModelStr, strlen(cacheModelStr)); STR_WITH_MAXSIZE_TO_VARSTR(cacheModelVstr, cacheModelStr, 24);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, rows, (const char *)cacheModelVstr, false); colDataAppend(pColInfo, rows, (const char *)cacheModelVstr, false);

14
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -102,7 +102,7 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
dnodeObj.updateTime = dnodeObj.createdTime; dnodeObj.updateTime = dnodeObj.createdTime;
dnodeObj.port = tsServerPort; dnodeObj.port = tsServerPort;
memcpy(&dnodeObj.fqdn, tsLocalFqdn, TSDB_FQDN_LEN); memcpy(&dnodeObj.fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port); snprintf(dnodeObj.ep, TSDB_EP_LEN - 1, "%s:%u", dnodeObj.fqdn, dnodeObj.port);
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL, "create-dnode"); pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL, "create-dnode");
if (pTrans == NULL) goto _OVER; if (pTrans == NULL) goto _OVER;
@ -190,7 +190,7 @@ _OVER:
static int32_t mndDnodeActionInsert(SSdb *pSdb, SDnodeObj *pDnode) { static int32_t mndDnodeActionInsert(SSdb *pSdb, SDnodeObj *pDnode) {
mTrace("dnode:%d, perform insert action, row:%p", pDnode->id, pDnode); mTrace("dnode:%d, perform insert action, row:%p", pDnode->id, pDnode);
pDnode->offlineReason = DND_REASON_STATUS_NOT_RECEIVED; pDnode->offlineReason = DND_REASON_STATUS_NOT_RECEIVED;
snprintf(pDnode->ep, TSDB_EP_LEN, "%s:%u", pDnode->fqdn, pDnode->port); snprintf(pDnode->ep, TSDB_EP_LEN - 1, "%s:%u", pDnode->fqdn, pDnode->port);
return 0; return 0;
} }
@ -253,7 +253,7 @@ int32_t mndGetDnodeSize(SMnode *pMnode) {
bool mndIsDnodeOnline(SDnodeObj *pDnode, int64_t curMs) { bool mndIsDnodeOnline(SDnodeObj *pDnode, int64_t curMs) {
int64_t interval = TABS(pDnode->lastAccessTime - curMs); int64_t interval = TABS(pDnode->lastAccessTime - curMs);
if (interval > 5000 * tsStatusInterval) { if (interval > 5000 * (int64_t)tsStatusInterval) {
if (pDnode->rebootTime > 0) { if (pDnode->rebootTime > 0) {
pDnode->offlineReason = DND_REASON_STATUS_MSG_TIMEOUT; pDnode->offlineReason = DND_REASON_STATUS_MSG_TIMEOUT;
} }
@ -486,7 +486,7 @@ static int32_t mndCreateDnode(SMnode *pMnode, SRpcMsg *pReq, SCreateDnodeReq *pC
dnodeObj.updateTime = dnodeObj.createdTime; dnodeObj.updateTime = dnodeObj.createdTime;
dnodeObj.port = pCreate->port; dnodeObj.port = pCreate->port;
memcpy(dnodeObj.fqdn, pCreate->fqdn, TSDB_FQDN_LEN); memcpy(dnodeObj.fqdn, pCreate->fqdn, TSDB_FQDN_LEN);
snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port); snprintf(dnodeObj.ep, TSDB_EP_LEN - 1, "%s:%u", dnodeObj.fqdn, dnodeObj.port);
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_GLOBAL, pReq, "create-dnode"); pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_GLOBAL, pReq, "create-dnode");
if (pTrans == NULL) goto _OVER; if (pTrans == NULL) goto _OVER;
@ -673,12 +673,14 @@ static int32_t mndDropDnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode, SM
mInfo("trans:%d, used to drop dnode:%d", pTrans->id, pDnode->id); mInfo("trans:%d, used to drop dnode:%d", pTrans->id, pDnode->id);
pRaw = mndDnodeActionEncode(pDnode); pRaw = mndDnodeActionEncode(pDnode);
if (pRaw == NULL || mndTransAppendRedolog(pTrans, pRaw) != 0) goto _OVER; if (pRaw == NULL) goto _OVER;
if (mndTransAppendRedolog(pTrans, pRaw) != 0) goto _OVER;
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPING); sdbSetRawStatus(pRaw, SDB_STATUS_DROPPING);
pRaw = NULL; pRaw = NULL;
pRaw = mndDnodeActionEncode(pDnode); pRaw = mndDnodeActionEncode(pDnode);
if (pRaw == NULL || mndTransAppendCommitlog(pTrans, pRaw) != 0) goto _OVER; if (pRaw == NULL) goto _OVER;
if (mndTransAppendCommitlog(pTrans, pRaw) != 0) goto _OVER;
sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED); sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
pRaw = NULL; pRaw = NULL;

9
source/dnode/mnode/impl/src/mndFunc.c
View File

@ -257,15 +257,18 @@ static int32_t mndDropFunc(SMnode *pMnode, SRpcMsg *pReq, SFuncObj *pFunc) {
mInfo("trans:%d, used to drop user:%s", pTrans->id, pFunc->name); mInfo("trans:%d, used to drop user:%s", pTrans->id, pFunc->name);
SSdbRaw *pRedoRaw = mndFuncActionEncode(pFunc); SSdbRaw *pRedoRaw = mndFuncActionEncode(pFunc);
if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) goto _OVER; if (pRedoRaw == NULL) goto _OVER;
if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) goto _OVER;
sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING); sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING);
SSdbRaw *pUndoRaw = mndFuncActionEncode(pFunc); SSdbRaw *pUndoRaw = mndFuncActionEncode(pFunc);
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) goto _OVER; if (pUndoRaw == NULL) goto _OVER;
if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) goto _OVER;
sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY); sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY);
SSdbRaw *pCommitRaw = mndFuncActionEncode(pFunc); SSdbRaw *pCommitRaw = mndFuncActionEncode(pFunc);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER; if (pCommitRaw == NULL) goto _OVER;
if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED); sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER; if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;

28
source/dnode/mnode/impl/src/mndGrant.c
View File

@ -30,85 +30,85 @@ static int32_t mndRetrieveGrant(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
cols = 0; cols = 0;
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols); SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
const char *src = "community"; const char *src = "community";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "false"; src = "false";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
cols++; cols++;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols); pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
src = "unlimited"; src = "unlimited";
STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); STR_WITH_MAXSIZE_TO_VARSTR(tmp, src, 32);
colDataAppend(pColInfo, numOfRows, tmp, false); colDataAppend(pColInfo, numOfRows, tmp, false);
numOfRows++; numOfRows++;

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

@ -649,7 +649,7 @@ int32_t mndProcessRpcMsg(SRpcMsg *pMsg) {
void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) { void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) {
tmsg_t type = TMSG_INDEX(msgType); tmsg_t type = TMSG_INDEX(msgType);
if (type >= 0 && type < TDMT_MAX) { if (type < TDMT_MAX) {
pMnode->msgFp[type] = fp; pMnode->msgFp[type] = fp;
} }
} }

11
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -93,6 +93,7 @@ static int32_t mndCreateDefaultMnode(SMnode *pMnode) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL, "create-mnode"); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL, "create-mnode");
if (pTrans == NULL) { if (pTrans == NULL) {
sdbFreeRaw(pRaw);
mError("mnode:%d, failed to create since %s", mnodeObj.id, terrstr()); mError("mnode:%d, failed to create since %s", mnodeObj.id, terrstr());
return -1; return -1;
} }
@ -220,8 +221,12 @@ bool mndIsMnode(SMnode *pMnode, int32_t dnodeId) {
void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) { void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE); int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE);
void *pIter = NULL; if (totalMnodes == 0) {
syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
return;
}
void *pIter = NULL;
while (1) { while (1) {
SMnodeObj *pObj = NULL; SMnodeObj *pObj = NULL;
pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj); pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj);
@ -658,7 +663,7 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
colDataAppend(pColInfo, numOfRows, (const char *)&pObj->id, false); colDataAppend(pColInfo, numOfRows, (const char *)&pObj->id, false);
char b1[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0}; char b1[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(b1, pObj->pDnode->ep, pShow->pMeta->pSchemas[cols].bytes); STR_WITH_MAXSIZE_TO_VARSTR(b1, pObj->pDnode->ep, TSDB_EP_LEN + VARSTR_HEADER_SIZE);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, b1, false); colDataAppend(pColInfo, numOfRows, b1, false);
@ -667,7 +672,7 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
if (pObj->id == pMnode->selfDnodeId) { if (pObj->id == pMnode->selfDnodeId) {
roles = syncStr(TAOS_SYNC_STATE_LEADER); roles = syncStr(TAOS_SYNC_STATE_LEADER);
} }
if (pObj->pDnode && mndIsDnodeOnline(pObj->pDnode, curMs)) { if (mndIsDnodeOnline(pObj->pDnode, curMs)) {
roles = syncStr(pObj->state); roles = syncStr(pObj->state);
if (pObj->state == TAOS_SYNC_STATE_LEADER && pObj->id != pMnode->selfDnodeId) { if (pObj->state == TAOS_SYNC_STATE_LEADER && pObj->id != pMnode->selfDnodeId) {
roles = syncStr(TAOS_SYNC_STATE_ERROR); roles = syncStr(TAOS_SYNC_STATE_ERROR);

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

@ -26,7 +26,7 @@ int32_t mndInitPerfsTableSchema(const SSysDbTableSchema *pSrc, int32_t colNum, S
} }
for (int32_t i = 0; i < colNum; ++i) { for (int32_t i = 0; i < colNum; ++i) {
strcpy(schema[i].name, pSrc[i].name); tstrncpy(schema[i].name, pSrc[i].name, sizeof(schema[i].name));
schema[i].type = pSrc[i].type; schema[i].type = pSrc[i].type;
schema[i].colId = i + 1; schema[i].colId = i + 1;
@ -40,7 +40,7 @@ int32_t mndInitPerfsTableSchema(const SSysDbTableSchema *pSrc, int32_t colNum, S
int32_t mndPerfsInitMeta(SHashObj *hash) { int32_t mndPerfsInitMeta(SHashObj *hash) {
STableMetaRsp meta = {0}; STableMetaRsp meta = {0};
strcpy(meta.dbFName, TSDB_INFORMATION_SCHEMA_DB); tstrncpy(meta.dbFName, TSDB_INFORMATION_SCHEMA_DB, sizeof(meta.dbFName));
meta.tableType = TSDB_SYSTEM_TABLE; meta.tableType = TSDB_SYSTEM_TABLE;
meta.sversion = 1; meta.sversion = 1;
meta.tversion = 1; meta.tversion = 1;
@ -50,7 +50,7 @@ int32_t mndPerfsInitMeta(SHashObj *hash) {
getPerfDbMeta(&pSysDbTableMeta, &size); getPerfDbMeta(&pSysDbTableMeta, &size);
for (int32_t i = 0; i < size; ++i) { for (int32_t i = 0; i < size; ++i) {
strcpy(meta.tbName, pSysDbTableMeta[i].name); tstrncpy(meta.tbName, pSysDbTableMeta[i].name, sizeof(meta.tbName));
meta.numOfColumns = pSysDbTableMeta[i].colNum; meta.numOfColumns = pSysDbTableMeta[i].colNum;
if (mndInitPerfsTableSchema(pSysDbTableMeta[i].schema, pSysDbTableMeta[i].colNum, &meta.pSchemas)) { if (mndInitPerfsTableSchema(pSysDbTableMeta[i].schema, pSysDbTableMeta[i].colNum, &meta.pSchemas)) {

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

@ -664,7 +664,7 @@ static int32_t mndProcessKillConnReq(SRpcMsg *pReq) {
static int32_t mndProcessSvrVerReq(SRpcMsg *pReq) { static int32_t mndProcessSvrVerReq(SRpcMsg *pReq) {
int32_t code = -1; int32_t code = -1;
SServerVerRsp rsp = {0}; SServerVerRsp rsp = {0};
strcpy(rsp.ver, version); tstrncpy(rsp.ver, version, sizeof(rsp.ver));
int32_t contLen = tSerializeSServerVerRsp(NULL, 0, &rsp); int32_t contLen = tSerializeSServerVerRsp(NULL, 0, &rsp);
if (contLen < 0) goto _over; if (contLen < 0) goto _over;
@ -702,7 +702,7 @@ static int32_t mndRetrieveConns(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
break; break;
} }
if ((taosGetTimestampMs() - pConn->lastAccessTimeMs) > (keepTime * 1000)) { if ((taosGetTimestampMs() - pConn->lastAccessTimeMs) > ((int64_t)keepTime * 1000)) {
continue; continue;
} }

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

@ -123,7 +123,7 @@ static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscri
req.subType = pSub->subType; req.subType = pSub->subType;
req.withMeta = pSub->withMeta; req.withMeta = pSub->withMeta;
req.suid = pSub->stbUid; req.suid = pSub->stbUid;
strncpy(req.subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN); tstrncpy(req.subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
int32_t tlen = sizeof(SMsgHead) + tEncodeSMqRebVgReq(NULL, &req); int32_t tlen = sizeof(SMsgHead) + tEncodeSMqRebVgReq(NULL, &req);
void *buf = taosMemoryMalloc(tlen); void *buf = taosMemoryMalloc(tlen);

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

@ -62,7 +62,9 @@ void vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot);
void vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId); void vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId);
int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen); int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen);
int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list); int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list);
int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list); int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list);
int32_t vnodeGetCtbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg), void *arg);
int32_t vnodeGetStbIdList(SVnode *pVnode, int64_t suid, SArray *list); int32_t vnodeGetStbIdList(SVnode *pVnode, int64_t suid, SArray *list);
void *vnodeGetIdx(SVnode *pVnode); void *vnodeGetIdx(SVnode *pVnode);
void *vnodeGetIvtIdx(SVnode *pVnode); void *vnodeGetIvtIdx(SVnode *pVnode);
@ -94,9 +96,12 @@ void metaReaderClear(SMetaReader *pReader);
int32_t metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid); int32_t metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid);
int metaGetTableEntryByName(SMetaReader *pReader, const char *name); int metaGetTableEntryByName(SMetaReader *pReader, const char *name);
int32_t metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags); 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); int32_t metaReadNext(SMetaReader *pReader);
const void *metaGetTableTagVal(void *tag, int16_t type, STagVal *tagVal); const void *metaGetTableTagVal(void *tag, int16_t type, STagVal *tagVal);
int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName); int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName);
int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid);
int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType);
bool metaIsTableExist(SMeta *pMeta, tb_uid_t uid); bool metaIsTableExist(SMeta *pMeta, tb_uid_t uid);
typedef struct SMetaFltParam { typedef struct SMetaFltParam {
@ -154,7 +159,7 @@ uint64_t getReaderMaxVersion(STsdbReader *pReader);
int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, SArray *pTableIdList, int32_t numOfCols, void **pReader); int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, SArray *pTableIdList, int32_t numOfCols, void **pReader);
int32_t tsdbRetrieveCacheRows(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, SArray *pTableUids); int32_t tsdbRetrieveCacheRows(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, SArray *pTableUids);
int32_t tsdbCacherowsReaderClose(void *pReader); void* tsdbCacherowsReaderClose(void *pReader);
int32_t tsdbGetTableSchema(SVnode *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid); int32_t tsdbGetTableSchema(SVnode *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid);
void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity); void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity);

7
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -110,6 +110,8 @@ int metaAlterTable(SMeta* pMeta, int64_t version, SVAlterTbReq* pReq
SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, bool isinline); SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, bool isinline);
STSchema* metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver); STSchema* metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
int32_t metaGetTbTSchemaEx(SMeta* pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema** ppTSchema); int32_t metaGetTbTSchemaEx(SMeta* pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema** ppTSchema);
int metaGetTableEntryByName(SMetaReader* pReader, const char* name);
tb_uid_t metaGetTableEntryUidByName(SMeta* pMeta, const char* name); tb_uid_t metaGetTableEntryUidByName(SMeta* pMeta, const char* name);
int64_t metaGetTbNum(SMeta* pMeta); int64_t metaGetTbNum(SMeta* pMeta);
int64_t metaGetTimeSeriesNum(SMeta* pMeta); int64_t metaGetTimeSeriesNum(SMeta* pMeta);
@ -183,8 +185,9 @@ int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg); int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema, SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema,
const char* stbFullName, SBatchDeleteReq* pDeleteReq); SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,
SBatchDeleteReq* pDeleteReq);
// sma // sma
int32_t smaInit(); int32_t smaInit();

12
source/dnode/vnode/src/meta/metaOpen.c
View File

@ -200,36 +200,30 @@ int metaClose(SMeta *pMeta) {
int32_t metaRLock(SMeta *pMeta) { int32_t metaRLock(SMeta *pMeta) {
int32_t ret = 0; int32_t ret = 0;
metaTrace("meta rlock %p B", &pMeta->lock); metaTrace("meta rlock %p", &pMeta->lock);
ret = taosThreadRwlockRdlock(&pMeta->lock); ret = taosThreadRwlockRdlock(&pMeta->lock);
metaTrace("meta rlock %p E", &pMeta->lock);
return ret; return ret;
} }
int32_t metaWLock(SMeta *pMeta) { int32_t metaWLock(SMeta *pMeta) {
int32_t ret = 0; int32_t ret = 0;
metaTrace("meta wlock %p B", &pMeta->lock); metaTrace("meta wlock %p", &pMeta->lock);
ret = taosThreadRwlockWrlock(&pMeta->lock); ret = taosThreadRwlockWrlock(&pMeta->lock);
metaTrace("meta wlock %p E", &pMeta->lock);
return ret; return ret;
} }
int32_t metaULock(SMeta *pMeta) { int32_t metaULock(SMeta *pMeta) {
int32_t ret = 0; int32_t ret = 0;
metaTrace("meta ulock %p B", &pMeta->lock); metaTrace("meta ulock %p", &pMeta->lock);
ret = taosThreadRwlockUnlock(&pMeta->lock); ret = taosThreadRwlockUnlock(&pMeta->lock);
metaTrace("meta ulock %p E", &pMeta->lock);
return ret; return ret;
} }

47
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -202,6 +202,38 @@ int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName) {
return 0; return 0;
} }
int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid) {
int code = 0;
SMetaReader mr = {0};
metaReaderInit(&mr, (SMeta *)meta, 0);
SMeta *pMeta = mr.pMeta;
SMetaReader *pReader = &mr;
// query name.idx
if (tdbTbGet(pMeta->pNameIdx, tbName, strlen(tbName) + 1, &pReader->pBuf, &pReader->szBuf) < 0) {
terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
metaReaderClear(&mr);
return -1;
}
*uid = *(tb_uid_t *)pReader->pBuf;
metaReaderClear(&mr);
return 0;
}
int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType) {
int code = 0;
SMetaReader mr = {0};
metaReaderInit(&mr, (SMeta *)meta, 0);
if (metaGetTableEntryByName(&mr, tbName) == 0) {
*tbType = mr.me.type;
}
metaReaderClear(&mr);
return 0;
}
int metaReadNext(SMetaReader *pReader) { int metaReadNext(SMetaReader *pReader) {
SMeta *pMeta = pReader->pMeta; SMeta *pMeta = pReader->pMeta;
@ -1102,6 +1134,21 @@ END:
return ret; return ret;
} }
int32_t metaGetTableTagsOpt(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags) {
int32_t sz = uidList ? taosArrayGetSize(uidList) : 0;
for (int i = 0; i < sz; i++) {
tb_uid_t *id = taosArrayGet(uidList, i);
SCtbIdxKey ctbIdxKey = {.suid = suid, .uid = *id};
void *val = NULL;
int32_t len = 0;
if (taosHashGet(tags, id, sizeof(tb_uid_t)) == NULL &&
0 == tdbTbGet(pMeta->pCtbIdx, &ctbIdxKey, sizeof(SCtbIdxKey), &val, &len)) {
taosHashPut(tags, id, sizeof(tb_uid_t), val, len);
}
}
return 0;
}
int32_t metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags) { int32_t metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags) {
SMCtbCursor *pCur = metaOpenCtbCursor(pMeta, suid); SMCtbCursor *pCur = metaOpenCtbCursor(pMeta, suid);

4
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -101,7 +101,7 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, SArray* pTableIdList
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t tsdbCacherowsReaderClose(void* pReader) { void* tsdbCacherowsReaderClose(void* pReader) {
SCacheRowsReader* p = pReader; SCacheRowsReader* p = pReader;
if (p->pSchema != NULL) { if (p->pSchema != NULL) {
@ -114,7 +114,7 @@ int32_t tsdbCacherowsReaderClose(void* pReader) {
} }
taosMemoryFree(pReader); taosMemoryFree(pReader);
return TSDB_CODE_SUCCESS; return NULL;
} }
static int32_t doExtractCacheRow(SCacheRowsReader* pr, SLRUCache* lruCache, uint64_t uid, STSRow** pRow, static int32_t doExtractCacheRow(SCacheRowsReader* pr, SLRUCache* lruCache, uint64_t uid, STSRow** pRow,

3
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -409,6 +409,9 @@ int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t vnodeGetCtbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg), void *arg) {
return 0;
}
int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list) { int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list) {
SMCtbCursor *pCur = metaOpenCtbCursor(pVnode->pMeta, suid); SMCtbCursor *pCur = metaOpenCtbCursor(pVnode->pMeta, suid);

10
source/libs/executor/inc/executil.h
View File

@ -15,7 +15,6 @@
#ifndef TDENGINE_QUERYUTIL_H #ifndef TDENGINE_QUERYUTIL_H
#define TDENGINE_QUERYUTIL_H #define TDENGINE_QUERYUTIL_H
#include "vnode.h"
#include "function.h" #include "function.h"
#include "nodes.h" #include "nodes.h"
#include "plannodes.h" #include "plannodes.h"
@ -23,6 +22,7 @@
#include "tcommon.h" #include "tcommon.h"
#include "tpagedbuf.h" #include "tpagedbuf.h"
#include "tsimplehash.h" #include "tsimplehash.h"
#include "vnode.h"
#define T_LONG_JMP(_obj, _c) \ #define T_LONG_JMP(_obj, _c) \
do { \ do { \
@ -101,11 +101,6 @@ static FORCE_INLINE SResultRow* getResultRowByPos(SDiskbasedBuf* pBuf, SResultRo
return pRow; return pRow;
} }
static FORCE_INLINE void setResultBufPageDirty(SDiskbasedBuf* pBuf, SResultRowPosition* pos) {
void* pPage = getBufPage(pBuf, pos->pageId);
setBufPageDirty(pPage, true);
}
void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order); void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order);
void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo); void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);
@ -117,7 +112,8 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);
SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode); SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode);
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext); EDealRes doTranslateTagExpr(SNode** pNode, void* pContext);
int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond, STableListInfo* pListInfo); int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
STableListInfo* pListInfo);
int32_t getGroupIdFromTagsVal(void* pMeta, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId); int32_t getGroupIdFromTagsVal(void* pMeta, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId);
int32_t getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableListInfo* pTableListInfo); int32_t getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableListInfo* pTableListInfo);
size_t getTableTagsBufLen(const SNodeList* pGroups); size_t getTableTagsBufLen(const SNodeList* pGroups);

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

@ -577,13 +577,7 @@ typedef struct SIntervalAggOperatorInfo {
int32_t inputOrder; // input data ts order int32_t inputOrder; // input data ts order
EOPTR_EXEC_MODEL execModel; // operator execution model [batch model|stream model] EOPTR_EXEC_MODEL execModel; // operator execution model [batch model|stream model]
STimeWindowAggSupp twAggSup; STimeWindowAggSupp twAggSup;
bool invertible;
SArray* pPrevValues; // SArray<SGroupKeys> used to keep the previous not null value for interpolation. SArray* pPrevValues; // SArray<SGroupKeys> used to keep the previous not null value for interpolation.
bool ignoreExpiredData;
SArray* pRecycledPages;
SArray* pDelWins; // SWinRes
int32_t delIndex;
SSDataBlock* pDelRes;
SNode* pCondition; SNode* pCondition;
} SIntervalAggOperatorInfo; } SIntervalAggOperatorInfo;
@ -609,38 +603,21 @@ typedef struct SStreamIntervalOperatorInfo {
STimeWindowAggSupp twAggSup; STimeWindowAggSupp twAggSup;
bool invertible; bool invertible;
bool ignoreExpiredData; bool ignoreExpiredData;
SArray* pRecycledPages;
SArray* pDelWins; // SWinRes SArray* pDelWins; // SWinRes
int32_t delIndex; int32_t delIndex;
SSDataBlock* pDelRes; SSDataBlock* pDelRes;
bool isFinal;
} SStreamIntervalOperatorInfo;
typedef struct SStreamFinalIntervalOperatorInfo {
// SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
SOptrBasicInfo binfo; // basic info
SAggSupporter aggSup; // aggregate supporter
SExprSupp scalarSupp; // supporter for perform scalar function
SGroupResInfo groupResInfo; // multiple results build supporter
SInterval interval; // interval info
int32_t primaryTsIndex; // primary time stamp slot id from result of downstream operator.
int32_t order; // current SSDataBlock scan order
STimeWindowAggSupp twAggSup;
SArray* pChildren;
SSDataBlock* pUpdateRes; SSDataBlock* pUpdateRes;
bool returnUpdate; bool returnUpdate;
SPhysiNode* pPhyNode; // create new child SPhysiNode* pPhyNode; // create new child
bool isFinal;
SHashObj* pPullDataMap; SHashObj* pPullDataMap;
SArray* pPullWins; // SPullWindowInfo SArray* pPullWins; // SPullWindowInfo
int32_t pullIndex; int32_t pullIndex;
SSDataBlock* pPullDataRes; SSDataBlock* pPullDataRes;
bool ignoreExpiredData; bool isFinal;
SArray* pRecycledPages; SArray* pChildren;
SArray* pDelWins; // SWinRes SStreamState* pState;
int32_t delIndex; SWinKey delKey;
SSDataBlock* pDelRes; } SStreamIntervalOperatorInfo;
} SStreamFinalIntervalOperatorInfo;
typedef struct SAggOperatorInfo { typedef struct SAggOperatorInfo {
// SOptrBasicInfo should be first, SAggSupporter should be second for stream encode // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
@ -1086,7 +1063,7 @@ bool functionNeedToExecute(SqlFunctionCtx* pCtx);
bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup); bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup); bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup); bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SOperatorInfo* pOperator, STimeWindowAggSupp* pTwSup); bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup);
void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, uint64_t* pGp); void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, uint64_t* pGp);
void printDataBlock(SSDataBlock* pBlock, const char* flag); void printDataBlock(SSDataBlock* pBlock, const char* flag);
uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId); uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
@ -1108,13 +1085,12 @@ void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsCol
bool groupbyTbname(SNodeList* pGroupList); bool groupbyTbname(SNodeList* pGroupList);
int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey); int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey);
void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput); void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput);
int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
SGroupResInfo* pGroupResInfo); SGroupResInfo* pGroupResInfo);
int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx, int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup);
SExecTaskInfo* pTaskInfo); int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
int32_t releaseOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult); int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
int32_t saveOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order); void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
#ifdef __cplusplus #ifdef __cplusplus

11
source/libs/executor/src/cachescanoperator.c
View File

@ -207,7 +207,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
} }
} }
tsdbCacherowsReaderClose(pInfo->pLastrowReader); pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
return pInfo->pRes; return pInfo->pRes;
} }
} }
@ -220,6 +220,15 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
void destroyLastrowScanOperator(void* param) { void destroyLastrowScanOperator(void* param) {
SLastrowScanInfo* pInfo = (SLastrowScanInfo*)param; SLastrowScanInfo* pInfo = (SLastrowScanInfo*)param;
blockDataDestroy(pInfo->pRes); blockDataDestroy(pInfo->pRes);
blockDataDestroy(pInfo->pBufferredRes);
taosMemoryFree(pInfo->pSlotIds);
taosArrayDestroy(pInfo->pUidList);
taosArrayDestroy(pInfo->pColMatchInfo);
if (pInfo->pLastrowReader != NULL) {
pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
}
taosMemoryFreeClear(param); taosMemoryFreeClear(param);
} }

101
source/libs/executor/src/executil.c
View File

@ -26,6 +26,9 @@
#include "executorimpl.h" #include "executorimpl.h"
#include "tcompression.h" #include "tcompression.h"
static int32_t optimizeTbnameInCond(void* metaHandle, int64_t suid, SArray* list, SNode* pTagCond);
static int32_t optimizeTbnameInCondImpl(void* metaHandle, int64_t suid, SArray* list, SNode* pTagCond);
void initResultRowInfo(SResultRowInfo* pResultRowInfo) { void initResultRowInfo(SResultRowInfo* pResultRowInfo) {
pResultRowInfo->size = 0; pResultRowInfo->size = 0;
pResultRowInfo->cur.pageId = -1; pResultRowInfo->cur.pageId = -1;
@ -407,12 +410,19 @@ static SColumnInfoData* getColInfoResult(void* metaHandle, uint64_t suid, SArray
// int64_t stt = taosGetTimestampUs(); // int64_t stt = taosGetTimestampUs();
tags = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK); tags = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
int32_t filter = optimizeTbnameInCond(metaHandle, suid, uidList, pTagCond);
if (filter == -1) {
code = metaGetTableTags(metaHandle, suid, uidList, tags); code = metaGetTableTags(metaHandle, suid, uidList, tags);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), suid); qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), suid);
terrno = code; terrno = code;
goto end; goto end;
} }
} else {
metaGetTableTagsOpt(metaHandle, suid, uidList, tags);
qInfo("succ to get table from meta idx, suid:%" PRIu64, suid);
}
int32_t rows = taosArrayGetSize(uidList); int32_t rows = taosArrayGetSize(uidList);
if (rows == 0) { if (rows == 0) {
@ -742,6 +752,94 @@ end:
return code; return code;
} }
static int tableUidCompare(const void* a, const void* b) {
uint64_t u1 = *(uint64_t*)a;
uint64_t u2 = *(uint64_t*)b;
if (u1 == u2) {
return 0;
}
return u1 < u2 ? -1 : 1;
}
static int32_t optimizeTbnameInCond(void* metaHandle, int64_t suid, SArray* list, SNode* cond) {
if (nodeType(cond) == QUERY_NODE_OPERATOR) {
return optimizeTbnameInCondImpl(metaHandle, suid, list, cond);
}
if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
return -1;
}
bool hasTbnameCond = false;
SLogicConditionNode* pNode = (SLogicConditionNode*)cond;
SNodeList* pList = (SNodeList*)pNode->pParameterList;
int32_t len = LIST_LENGTH(pList);
if (len <= 0) return -1;
SListCell* cell = pList->pHead;
for (int i = 0; i < len; i++) {
if (cell == NULL) break;
if (optimizeTbnameInCondImpl(metaHandle, suid, list, cell->pNode) == 0) {
hasTbnameCond = true;
}
cell = cell->pNext;
}
taosArraySort(list, tableUidCompare);
taosArrayRemoveDuplicate(list, tableUidCompare, NULL);
return hasTbnameCond == true ? 0 : -1;
}
static int32_t optimizeTbnameInCondImpl(void* metaHandle, int64_t suid, SArray* list, SNode* pTagCond) {
if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {
return -1;
}
SOperatorNode* pNode = (SOperatorNode*)pTagCond;
if (pNode->opType != OP_TYPE_IN) {
return -1;
}
if ((pNode->pLeft != NULL && nodeType(pNode->pLeft) == QUERY_NODE_COLUMN &&
((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME) &&
(pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {
SNodeListNode* pList = (SNodeListNode*)pNode->pRight;
int32_t len = LIST_LENGTH(pList->pNodeList);
if (len <= 0) return -1;
SListCell* cell = pList->pNodeList->pHead;
SArray* pTbList = taosArrayInit(len, sizeof(void*));
for (int i = 0; i < pList->pNodeList->length; i++) {
SValueNode* valueNode = (SValueNode*)cell->pNode;
if (!IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {
taosArrayDestroy(pTbList);
return -1;
}
char* name = varDataVal(valueNode->datum.p);
taosArrayPush(pTbList, &name);
cell = cell->pNext;
}
for (int i = 0; i < taosArrayGetSize(pTbList); i++) {
char* name = taosArrayGetP(pTbList, i);
uint64_t uid = 0;
if (metaGetTableUidByName(metaHandle, name, &uid) == 0) {
ETableType tbType = TSDB_TABLE_MAX;
if (metaGetTableTypeByName(metaHandle, name, &tbType) == 0 && tbType == TSDB_CHILD_TABLE) {
taosArrayPush(list, &uid);
} else {
taosArrayDestroy(pTbList);
return -1;
}
} else {
qWarn("failed to get tableIds from by table name: %s, reason: %s", name, tstrerror(terrno));
terrno = 0;
}
}
taosArrayDestroy(pTbList);
return 0;
}
return -1;
}
int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond, int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
STableListInfo* pListInfo) { STableListInfo* pListInfo) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
@ -767,9 +865,6 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode,
qError("failed to get tableIds from index, reason:%s, suid:%" PRIu64, tstrerror(code), tableUid); qError("failed to get tableIds from index, reason:%s, suid:%" PRIu64, tstrerror(code), tableUid);
code = TDB_CODE_SUCCESS; code = TDB_CODE_SUCCESS;
} }
// int64_t stt1 = taosGetTimestampUs();
// qDebug("generate table list, cost:%ld us", stt1-stt);
} else if (!pTagCond) { } else if (!pTagCond) {
vnodeGetCtbIdList(pVnode, pScanNode->suid, res); vnodeGetCtbIdList(pVnode, pScanNode->suid, res);
} }

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

@ -4183,9 +4183,8 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlF
return code; return code;
} }
int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx, int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup) {
SExecTaskInfo* pTaskInfo) {
SWinKey key = { SWinKey key = {
.ts = win->skey, .ts = win->skey,
.groupId = tableGroupId, .groupId = tableGroupId,
@ -4194,7 +4193,7 @@ int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupI
int32_t size = pAggSup->resultRowSize; int32_t size = pAggSup->resultRowSize;
tSimpleHashPut(pAggSup->pResultRowHashTable, &key, sizeof(SWinKey), NULL, 0); tSimpleHashPut(pAggSup->pResultRowHashTable, &key, sizeof(SWinKey), NULL, 0);
if (streamStateAddIfNotExist(pTaskInfo->streamInfo.pState, &key, (void**)&value, &size) < 0) { if (streamStateAddIfNotExist(pState, &key, (void**)&value, &size) < 0) {
return TSDB_CODE_QRY_OUT_OF_MEMORY; return TSDB_CODE_QRY_OUT_OF_MEMORY;
} }
*pResult = (SResultRow*)value; *pResult = (SResultRow*)value;
@ -4205,18 +4204,17 @@ int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupI
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t releaseOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult) { int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult) {
streamStateReleaseBuf(pTaskInfo->streamInfo.pState, pKey, pResult); streamStateReleaseBuf(pState, pKey, pResult);
/*taosMemoryFree((*(void**)pResult));*/
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t saveOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult, int32_t resSize) { int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize) {
streamStatePut(pTaskInfo->streamInfo.pState, pKey, pResult, resSize); streamStatePut(pState, pKey, pResult, resSize);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
SGroupResInfo* pGroupResInfo) { SGroupResInfo* pGroupResInfo) {
SExprInfo* pExprInfo = pSup->pExprInfo; SExprInfo* pExprInfo = pSup->pExprInfo;
int32_t numOfExprs = pSup->numOfExprs; int32_t numOfExprs = pSup->numOfExprs;
@ -4233,14 +4231,14 @@ int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock
.ts = *(TSKEY*)pPos->key, .ts = *(TSKEY*)pPos->key,
.groupId = pPos->groupId, .groupId = pPos->groupId,
}; };
int32_t code = streamStateGet(pTaskInfo->streamInfo.pState, &key, &pVal, &size); int32_t code = streamStateGet(pState, &key, &pVal, &size);
ASSERT(code == 0); ASSERT(code == 0);
SResultRow* pRow = (SResultRow*)pVal; SResultRow* pRow = (SResultRow*)pVal;
doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset); doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset);
// no results, continue to check the next one // no results, continue to check the next one
if (pRow->numOfRows == 0) { if (pRow->numOfRows == 0) {
pGroupResInfo->index += 1; pGroupResInfo->index += 1;
releaseOutputBuf(pTaskInfo, &key, pRow); releaseOutputBuf(pState, &key, pRow);
continue; continue;
} }
@ -4249,14 +4247,14 @@ int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock
} else { } else {
// current value belongs to different group, it can't be packed into one datablock // current value belongs to different group, it can't be packed into one datablock
if (pBlock->info.groupId != pPos->groupId) { if (pBlock->info.groupId != pPos->groupId) {
releaseOutputBuf(pTaskInfo, &key, pRow); releaseOutputBuf(pState, &key, pRow);
break; break;
} }
} }
if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) { if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
ASSERT(pBlock->info.rows > 0); ASSERT(pBlock->info.rows > 0);
releaseOutputBuf(pTaskInfo, &key, pRow); releaseOutputBuf(pState, &key, pRow);
break; break;
} }
@ -4286,7 +4284,7 @@ int32_t buildDataBlockFromGroupRes(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock
} }
pBlock->info.rows += pRow->numOfRows; pBlock->info.rows += pRow->numOfRows;
releaseOutputBuf(pTaskInfo, &key, pRow); releaseOutputBuf(pState, &key, pRow);
} }
blockDataUpdateTsWindow(pBlock, 0); blockDataUpdateTsWindow(pBlock, 0);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;

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

@ -1357,7 +1357,8 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
// must check update info first. // must check update info first.
bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]); bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]);
bool closedWin = isClosed && isSignleIntervalWindow(pInfo) && bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&
isDeletedStreamWindow(&win, pBlock->info.groupId, pInfo->pTableScanOp, &pInfo->twAggSup); isDeletedStreamWindow(&win, pBlock->info.groupId,
pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, &pInfo->twAggSup);
if ((update || closedWin) && out) { if ((update || closedWin) && out) {
qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin); qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);
uint64_t gpId = 0; uint64_t gpId = 0;
@ -2135,6 +2136,9 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
pInfo->pUpdateInfo = NULL; pInfo->pUpdateInfo = NULL;
pInfo->pTableScanOp = pTableScanOp; pInfo->pTableScanOp = pTableScanOp;
if (pInfo->pTableScanOp->pTaskInfo->streamInfo.pState) {
streamStateSetNumber(pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, -1);
}
pInfo->readHandle = *pHandle; pInfo->readHandle = *pHandle;
pInfo->tableUid = pScanPhyNode->uid; pInfo->tableUid = pScanPhyNode->uid;

4
source/libs/executor/src/tfill.c
View File

@ -1620,9 +1620,7 @@ SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFi
goto _error; goto _error;
} }
SInterval* pInterval = QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL == downstream->operatorType SInterval* pInterval = &((SStreamIntervalOperatorInfo*)downstream->info)->interval;
? &((SStreamFinalIntervalOperatorInfo*)downstream->info)->interval
: &((SStreamIntervalOperatorInfo*)downstream->info)->interval;
int32_t numOfFillCols = 0; int32_t numOfFillCols = 0;
SExprInfo* pFillExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &numOfFillCols); SExprInfo* pFillExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &numOfFillCols);
pInfo->pFillSup = initStreamFillSup(pPhyFillNode, pInterval, pFillExprInfo, numOfFillCols); pInfo->pFillSup = initStreamFillSup(pPhyFillNode, pInterval, pFillExprInfo, numOfFillCols);

703
source/libs/executor/src/timewindowoperator.c
View File

File diff suppressed because it is too large Load Diff

12
source/libs/index/src/index.c
View File

@ -54,11 +54,17 @@
void* indexQhandle = NULL; void* indexQhandle = NULL;
int32_t indexRefMgt; int32_t indexRefMgt;
int32_t indexThreads = 5;
static void indexDestroy(void* sIdx); static void indexDestroy(void* sIdx);
void indexInit() { void indexInit(int32_t threadNum) {
indexThreads = threadNum;
if (indexThreads <= 1) indexThreads = INDEX_NUM_OF_THREADS;
}
void indexEnvInit() {
// refactor later // refactor later
indexQhandle = taosInitScheduler(INDEX_QUEUE_SIZE, INDEX_NUM_OF_THREADS, "index", NULL); indexQhandle = taosInitScheduler(INDEX_QUEUE_SIZE, indexThreads, "index", NULL);
indexRefMgt = taosOpenRef(1000, indexDestroy); indexRefMgt = taosOpenRef(1000, indexDestroy);
} }
void indexCleanup() { void indexCleanup() {
@ -99,7 +105,7 @@ static void indexWait(void* idx) {
int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) { int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
int ret = TSDB_CODE_SUCCESS; int ret = TSDB_CODE_SUCCESS;
taosThreadOnce(&isInit, indexInit); taosThreadOnce(&isInit, indexEnvInit);
SIndex* idx = taosMemoryCalloc(1, sizeof(SIndex)); SIndex* idx = taosMemoryCalloc(1, sizeof(SIndex));
if (idx == NULL) { if (idx == NULL) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;

2
source/libs/index/src/indexCache.c
View File

@ -22,7 +22,7 @@
#define MAX_INDEX_KEY_LEN 256 // test only, change later #define MAX_INDEX_KEY_LEN 256 // test only, change later
#define MEM_TERM_LIMIT 10 * 10000 #define MEM_TERM_LIMIT 10 * 10000
#define MEM_THRESHOLD 512 * 1024 #define MEM_THRESHOLD 8 * 512 * 1024 // 8M
#define MEM_SIGNAL_QUIT MEM_THRESHOLD * 20 #define MEM_SIGNAL_QUIT MEM_THRESHOLD * 20
#define MEM_ESTIMATE_RADIO 1.5 #define MEM_ESTIMATE_RADIO 1.5

6
source/libs/index/src/indexTfile.c
View File

@ -555,6 +555,7 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
taosArraySort(v->tableId, idxUidCompare); taosArraySort(v->tableId, idxUidCompare);
taosArrayRemoveDuplicate(v->tableId, idxUidCompare, NULL); taosArrayRemoveDuplicate(v->tableId, idxUidCompare, NULL);
int32_t tbsz = taosArrayGetSize(v->tableId); int32_t tbsz = taosArrayGetSize(v->tableId);
if (tbsz == 0) continue;
fstOffset += TF_TABLE_TATOAL_SIZE(tbsz); fstOffset += TF_TABLE_TATOAL_SIZE(tbsz);
} }
tfileWriteFstOffset(tw, fstOffset); tfileWriteFstOffset(tw, fstOffset);
@ -566,6 +567,7 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
TFileValue* v = taosArrayGetP((SArray*)data, i); TFileValue* v = taosArrayGetP((SArray*)data, i);
int32_t tbsz = taosArrayGetSize(v->tableId); int32_t tbsz = taosArrayGetSize(v->tableId);
if (tbsz == 0) continue;
// check buf has enough space or not // check buf has enough space or not
int32_t ttsz = TF_TABLE_TATOAL_SIZE(tbsz); int32_t ttsz = TF_TABLE_TATOAL_SIZE(tbsz);
@ -592,6 +594,10 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
for (size_t i = 0; i < sz; i++) { for (size_t i = 0; i < sz; i++) {
// TODO, fst batch write later // TODO, fst batch write later
TFileValue* v = taosArrayGetP((SArray*)data, i); TFileValue* v = taosArrayGetP((SArray*)data, i);
int32_t tbsz = taosArrayGetSize(v->tableId);
if (tbsz == 0) continue;
if (tfileWriteData(tw, v) != 0) { if (tfileWriteData(tw, v) != 0) {
indexError("failed to write data: %s, offset: %d len: %d", v->colVal, v->offset, indexError("failed to write data: %s, offset: %d len: %d", v->colVal, v->offset,
(int)taosArrayGetSize(v->tableId)); (int)taosArrayGetSize(v->tableId));

130
source/libs/stream/src/streamState.c
View File

@ -18,6 +18,37 @@
#include "tcommon.h" #include "tcommon.h"
#include "ttimer.h" #include "ttimer.h"
// todo refactor
typedef struct SStateKey {
SWinKey key;
int64_t opNum;
} SStateKey;
static inline int SStateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
SStateKey* pWin1 = (SStateKey*)pKey1;
SStateKey* pWin2 = (SStateKey*)pKey2;
if (pWin1->opNum > pWin2->opNum) {
return 1;
} else if (pWin1->opNum < pWin2->opNum) {
return -1;
}
if (pWin1->key.ts > pWin2->key.ts) {
return 1;
} else if (pWin1->key.ts < pWin2->key.ts) {
return -1;
}
if (pWin1->key.groupId > pWin2->key.groupId) {
return 1;
} else if (pWin1->key.groupId < pWin2->key.groupId) {
return -1;
}
return 0;
}
SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) { SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) {
SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState)); SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState));
if (pState == NULL) { if (pState == NULL) {
@ -36,7 +67,7 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) {
} }
// open state storage backend // open state storage backend
if (tdbTbOpen("state.db", sizeof(SWinKey), -1, SWinKeyCmpr, pState->db, &pState->pStateDb) < 0) { if (tdbTbOpen("state.db", sizeof(SStateKey), -1, SStateKeyCmpr, pState->db, &pState->pStateDb) < 0) {
goto _err; goto _err;
} }
@ -130,8 +161,10 @@ int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key) {
return tdbTbDelete(pState->pFuncStateDb, key, sizeof(STupleKey), &pState->txn); return tdbTbDelete(pState->pFuncStateDb, key, sizeof(STupleKey), &pState->txn);
} }
// todo refactor
int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) { int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
return tdbTbUpsert(pState->pStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn); SStateKey sKey = {.key = *key, .opNum = pState->number};
return tdbTbUpsert(pState->pStateDb, &sKey, sizeof(SStateKey), value, vLen, &pState->txn);
} }
// todo refactor // todo refactor
@ -139,8 +172,10 @@ int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void*
return tdbTbUpsert(pState->pFillStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn); return tdbTbUpsert(pState->pFillStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn);
} }
// todo refactor
int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) { int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) {
return tdbTbGet(pState->pStateDb, key, sizeof(SWinKey), pVal, pVLen); SStateKey sKey = {.key = *key, .opNum = pState->number};
return tdbTbGet(pState->pStateDb, &sKey, sizeof(SStateKey), pVal, pVLen);
} }
// todo refactor // todo refactor
@ -148,10 +183,30 @@ int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal
return tdbTbGet(pState->pFillStateDb, key, sizeof(SWinKey), pVal, pVLen); return tdbTbGet(pState->pFillStateDb, key, sizeof(SWinKey), pVal, pVLen);
} }
// todo refactor
int32_t streamStateDel(SStreamState* pState, const SWinKey* key) { int32_t streamStateDel(SStreamState* pState, const SWinKey* key) {
return tdbTbDelete(pState->pStateDb, key, sizeof(SWinKey), &pState->txn); SStateKey sKey = {.key = *key, .opNum = pState->number};
return tdbTbDelete(pState->pStateDb, &sKey, sizeof(SStateKey), &pState->txn);
} }
int32_t streamStateClear(SStreamState* pState) {
SWinKey key = {.ts = 0, .groupId = 0};
streamStatePut(pState, &key, NULL, 0);
while (1) {
SStreamStateCur* pCur = streamStateSeekKeyNext(pState, &key);
SWinKey delKey = {0};
int32_t code = streamStateGetKVByCur(pCur, &delKey, NULL, 0);
if (code == 0) {
streamStateDel(pState, &delKey);
} else {
break;
}
}
return 0;
}
void streamStateSetNumber(SStreamState* pState, int32_t number) { pState->number = number; }
// todo refactor // todo refactor
int32_t streamStateFillDel(SStreamState* pState, const SWinKey* key) { int32_t streamStateFillDel(SStreamState* pState, const SWinKey* key) {
return tdbTbDelete(pState->pFillStateDb, key, sizeof(SWinKey), &pState->txn); return tdbTbDelete(pState->pFillStateDb, key, sizeof(SWinKey), &pState->txn);
@ -180,11 +235,13 @@ SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key) {
tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL); tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL);
int32_t c; int32_t c;
SStateKey sKey = {.key = *key, .opNum = pState->number};
tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c); tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c);
if (c != 0) { if (c != 0) {
taosMemoryFree(pCur); taosMemoryFree(pCur);
return NULL; return NULL;
} }
pCur->number = pState->number;
return pCur; return pCur;
} }
@ -214,6 +271,25 @@ SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key) {
} }
int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) { int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
if (!pCur) {
return -1;
}
const SStateKey* pKTmp = NULL;
int32_t kLen;
if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
return -1;
}
if (pKTmp->opNum != pCur->number) {
return -1;
}
*pKey = pKTmp->key;
return 0;
}
int32_t streamStateFillGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
if (!pCur) {
return -1;
}
const SWinKey* pKTmp = NULL; const SWinKey* pKTmp = NULL;
int32_t kLen; int32_t kLen;
if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) { if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
@ -225,7 +301,7 @@ int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void**
int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) { int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
uint64_t groupId = pKey->groupId; uint64_t groupId = pKey->groupId;
int32_t code = streamStateGetKVByCur(pCur, pKey, pVal, pVLen); int32_t code = streamStateFillGetKVByCur(pCur, pKey, pVal, pVLen);
if (code == 0) { if (code == 0) {
if (pKey->groupId == groupId) { if (pKey->groupId == groupId) {
return 0; return 0;
@ -234,6 +310,16 @@ int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const v
return -1; return -1;
} }
int32_t streamStateGetFirst(SStreamState* pState, SWinKey* key) {
// todo refactor
SWinKey tmp = {.ts = 0, .groupId = 0};
streamStatePut(pState, &tmp, NULL, 0);
SStreamStateCur* pCur = streamStateSeekKeyNext(pState, &tmp);
int32_t code = streamStateGetKVByCur(pCur, key, NULL, 0);
streamStateDel(pState, &tmp);
return code;
}
int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur) { int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur) {
// //
return tdbTbcMoveToFirst(pCur->pCur); return tdbTbcMoveToFirst(pCur->pCur);
@ -244,6 +330,34 @@ int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur) {
return tdbTbcMoveToLast(pCur->pCur); return tdbTbcMoveToLast(pCur->pCur);
} }
SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) {
return NULL;
}
pCur->number = pState->number;
if (tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL) < 0) {
taosMemoryFree(pCur);
return NULL;
}
SStateKey sKey = {.key = *key, .opNum = pState->number};
int32_t c;
if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateKey), &c) < 0) {
tdbTbcClose(pCur->pCur);
taosMemoryFree(pCur);
return NULL;
}
if (c > 0) return pCur;
if (tdbTbcMoveToNext(pCur->pCur) < 0) {
taosMemoryFree(pCur);
return NULL;
}
return pCur;
}
SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key) { SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur)); SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) { if (pCur == NULL) {
@ -303,9 +417,15 @@ int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur) {
int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) { int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) {
// //
if (!pCur) {
return -1;
}
return tdbTbcMoveToPrev(pCur->pCur); return tdbTbcMoveToPrev(pCur->pCur);
} }
void streamStateFreeCur(SStreamStateCur* pCur) { void streamStateFreeCur(SStreamStateCur* pCur) {
if (!pCur) {
return;
}
tdbTbcClose(pCur->pCur); tdbTbcClose(pCur->pCur);
taosMemoryFree(pCur); taosMemoryFree(pCur);
} }

2
source/libs/transport/src/transComm.c
View File

@ -193,7 +193,7 @@ bool transReadComplete(SConnBuffer* connBuf) {
int transSetConnOption(uv_tcp_t* stream) { int transSetConnOption(uv_tcp_t* stream) {
uv_tcp_nodelay(stream, 1); uv_tcp_nodelay(stream, 1);
int ret = uv_tcp_keepalive(stream, 5, 5); int ret = uv_tcp_keepalive(stream, 5, 60);
return ret; return ret;
} }

3
tests/script/tsim/stream/distributeInterval0.sim
View File

@ -33,7 +33,8 @@ if $data(2)[4] != ready then
endi endi
print ===== step2 print ===== step2
sql drop stream if exists stream_t1;
sql drop database if exists test;
sql create database test vgroups 4; sql create database test vgroups 4;
sql use test; sql use test;
sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int); sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int);