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 fix/TD-30837

This commit is contained in:
54liuyao 2024-10-30 10:09:48 +08:00
parent a0818d2a2d 5e08699fa6
commit d065d71821
47 changed files with 581 additions and 239 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
docs/en/14-reference/03-taos-sql/21-node.md
View File

@ -27,11 +27,15 @@ The preceding SQL command shows all dnodes in the cluster with the ID, endpoint,
## Delete a DNODE ## Delete a DNODE
```sql ```sql
DROP DNODE dnode_id DROP DNODE dnode_id [force] [unsafe]
``` ```
Note that deleting a dnode does not stop its process. You must stop the process after the dnode is deleted. Note that deleting a dnode does not stop its process. You must stop the process after the dnode is deleted.
Only online node is allowed to be deleted. Drop is executed with force option if the offline node need to be deleted.
Drop is executed with unsafe option if the node with single replica is offline, and the data on it is not able to be restored.
## Modify Dnode Configuration ## Modify Dnode Configuration
```sql ```sql

15
docs/examples/java/src/main/java/com/taos/example/ConsumerLoopFull.java
View File

@ -1,8 +1,9 @@
package com.taos.example; package com.taos.example;
import com.alibaba.fastjson.JSON; import com.fasterxml.jackson.core.JsonProcessingException;
import com.taosdata.jdbc.TSDBDriver; import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.tmq.*; import com.taosdata.jdbc.tmq.*;
import com.taosdata.jdbc.utils.JsonUtil;
import java.sql.*; import java.sql.*;
import java.time.Duration; import java.time.Duration;
@ -60,7 +61,7 @@ public class ConsumerLoopFull {
// ANCHOR_END: create_consumer // ANCHOR_END: create_consumer
} }
public static void pollExample(TaosConsumer<ResultBean> consumer) throws SQLException { public static void pollExample(TaosConsumer<ResultBean> consumer) throws SQLException, JsonProcessingException {
// ANCHOR: poll_data_code_piece // ANCHOR: poll_data_code_piece
List<String> topics = Collections.singletonList("topic_meters"); List<String> topics = Collections.singletonList("topic_meters");
try { try {
@ -73,7 +74,7 @@ public class ConsumerLoopFull {
for (ConsumerRecord<ResultBean> record : records) { for (ConsumerRecord<ResultBean> record : records) {
ResultBean bean = record.value(); ResultBean bean = record.value();
// Add your data processing logic here // Add your data processing logic here
System.out.println("data: " + JSON.toJSONString(bean)); System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(bean));
} }
} }
} catch (Exception ex) { } catch (Exception ex) {
@ -91,7 +92,7 @@ public class ConsumerLoopFull {
// ANCHOR_END: poll_data_code_piece // ANCHOR_END: poll_data_code_piece
} }
public static void seekExample(TaosConsumer<ResultBean> consumer) throws SQLException { public static void seekExample(TaosConsumer<ResultBean> consumer) throws SQLException, JsonProcessingException {
// ANCHOR: consumer_seek // ANCHOR: consumer_seek
List<String> topics = Collections.singletonList("topic_meters"); List<String> topics = Collections.singletonList("topic_meters");
try { try {
@ -99,7 +100,7 @@ public class ConsumerLoopFull {
consumer.subscribe(topics); consumer.subscribe(topics);
System.out.println("Subscribe topics successfully."); System.out.println("Subscribe topics successfully.");
Set<TopicPartition> assignment = consumer.assignment(); Set<TopicPartition> assignment = consumer.assignment();
System.out.println("Now assignment: " + JSON.toJSONString(assignment)); System.out.println("Now assignment: " + JsonUtil.getObjectMapper().writeValueAsString(assignment));
ConsumerRecords<ResultBean> records = ConsumerRecords.emptyRecord(); ConsumerRecords<ResultBean> records = ConsumerRecords.emptyRecord();
// make sure we have got some data // make sure we have got some data
@ -125,7 +126,7 @@ public class ConsumerLoopFull {
} }
public static void commitExample(TaosConsumer<ResultBean> consumer) throws SQLException { public static void commitExample(TaosConsumer<ResultBean> consumer) throws SQLException, JsonProcessingException {
// ANCHOR: commit_code_piece // ANCHOR: commit_code_piece
List<String> topics = Collections.singletonList("topic_meters"); List<String> topics = Collections.singletonList("topic_meters");
try { try {
@ -135,7 +136,7 @@ public class ConsumerLoopFull {
for (ConsumerRecord<ResultBean> record : records) { for (ConsumerRecord<ResultBean> record : records) {
ResultBean bean = record.value(); ResultBean bean = record.value();
// Add your data processing logic here // Add your data processing logic here
System.out.println("data: " + JSON.toJSONString(bean)); System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(bean));
} }
if (!records.isEmpty()) { if (!records.isEmpty()) {
// after processing the data, commit the offset manually // after processing the data, commit the offset manually

8
docs/examples/java/src/main/java/com/taos/example/ConsumerLoopImp.java
View File

@ -1,7 +1,7 @@
package com.taos.example; package com.taos.example;
import com.alibaba.fastjson.JSON;
import com.taosdata.jdbc.TSDBDriver; import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.utils.JsonUtil;
import java.sql.Connection; import java.sql.Connection;
import java.sql.DriverManager; import java.sql.DriverManager;
@ -31,7 +31,11 @@ public class ConsumerLoopImp {
final AbsConsumerLoop consumerLoop = new AbsConsumerLoop() { final AbsConsumerLoop consumerLoop = new AbsConsumerLoop() {
@Override @Override
public void process(ResultBean result) { public void process(ResultBean result) {
System.out.println("data: " + JSON.toJSONString(result)); try{
System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(result));
} catch (Exception e) {
throw new RuntimeException(e);
}
} }
}; };

15
docs/examples/java/src/main/java/com/taos/example/WsConsumerLoopFull.java
View File

@ -1,8 +1,9 @@
package com.taos.example; package com.taos.example;
import com.alibaba.fastjson.JSON; import com.fasterxml.jackson.core.JsonProcessingException;
import com.taosdata.jdbc.TSDBDriver; import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.tmq.*; import com.taosdata.jdbc.tmq.*;
import com.taosdata.jdbc.utils.JsonUtil;
import java.sql.*; import java.sql.*;
import java.time.Duration; import java.time.Duration;
@ -60,7 +61,7 @@ public class WsConsumerLoopFull {
// ANCHOR_END: create_consumer // ANCHOR_END: create_consumer
} }
public static void pollExample(TaosConsumer<ResultBean> consumer) throws SQLException { public static void pollExample(TaosConsumer<ResultBean> consumer) throws SQLException, JsonProcessingException {
// ANCHOR: poll_data_code_piece // ANCHOR: poll_data_code_piece
List<String> topics = Collections.singletonList("topic_meters"); List<String> topics = Collections.singletonList("topic_meters");
try { try {
@ -73,7 +74,7 @@ public class WsConsumerLoopFull {
for (ConsumerRecord<ResultBean> record : records) { for (ConsumerRecord<ResultBean> record : records) {
ResultBean bean = record.value(); ResultBean bean = record.value();
// Add your data processing logic here // Add your data processing logic here
System.out.println("data: " + JSON.toJSONString(bean)); System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(bean));
} }
} }
} catch (Exception ex) { } catch (Exception ex) {
@ -91,7 +92,7 @@ public class WsConsumerLoopFull {
// ANCHOR_END: poll_data_code_piece // ANCHOR_END: poll_data_code_piece
} }
public static void seekExample(TaosConsumer<ResultBean> consumer) throws SQLException { public static void seekExample(TaosConsumer<ResultBean> consumer) throws SQLException, JsonProcessingException {
// ANCHOR: consumer_seek // ANCHOR: consumer_seek
List<String> topics = Collections.singletonList("topic_meters"); List<String> topics = Collections.singletonList("topic_meters");
try { try {
@ -99,7 +100,7 @@ public class WsConsumerLoopFull {
consumer.subscribe(topics); consumer.subscribe(topics);
System.out.println("Subscribe topics successfully."); System.out.println("Subscribe topics successfully.");
Set<TopicPartition> assignment = consumer.assignment(); Set<TopicPartition> assignment = consumer.assignment();
System.out.println("Now assignment: " + JSON.toJSONString(assignment)); System.out.println("Now assignment: " + JsonUtil.getObjectMapper().writeValueAsString(assignment));
ConsumerRecords<ResultBean> records = ConsumerRecords.emptyRecord(); ConsumerRecords<ResultBean> records = ConsumerRecords.emptyRecord();
// make sure we have got some data // make sure we have got some data
@ -125,7 +126,7 @@ public class WsConsumerLoopFull {
} }
public static void commitExample(TaosConsumer<ResultBean> consumer) throws SQLException { public static void commitExample(TaosConsumer<ResultBean> consumer) throws SQLException, JsonProcessingException {
// ANCHOR: commit_code_piece // ANCHOR: commit_code_piece
List<String> topics = Collections.singletonList("topic_meters"); List<String> topics = Collections.singletonList("topic_meters");
try { try {
@ -135,7 +136,7 @@ public class WsConsumerLoopFull {
for (ConsumerRecord<ResultBean> record : records) { for (ConsumerRecord<ResultBean> record : records) {
ResultBean bean = record.value(); ResultBean bean = record.value();
// Add your data processing logic here // Add your data processing logic here
System.out.println("data: " + JSON.toJSONString(bean)); System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(bean));
} }
if (!records.isEmpty()) { if (!records.isEmpty()) {
// after processing the data, commit the offset manually // after processing the data, commit the offset manually

8
docs/examples/java/src/main/java/com/taos/example/WsConsumerLoopImp.java
View File

@ -1,7 +1,7 @@
package com.taos.example; package com.taos.example;
import com.alibaba.fastjson.JSON;
import com.taosdata.jdbc.TSDBDriver; import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.utils.JsonUtil;
import java.sql.Connection; import java.sql.Connection;
import java.sql.DriverManager; import java.sql.DriverManager;
@ -28,7 +28,11 @@ public abstract class WsConsumerLoopImp {
final AbsConsumerLoop consumerLoop = new AbsConsumerLoop() { final AbsConsumerLoop consumerLoop = new AbsConsumerLoop() {
@Override @Override
public void process(ResultBean result) { public void process(ResultBean result) {
System.out.println("data: " + JSON.toJSONString(result)); try{
System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(result));
} catch (Exception e) {
throw new RuntimeException(e);
}
} }
}; };

3
docs/examples/java/src/main/java/com/taos/example/highvolume/DataBaseMonitor.java
View File

@ -13,6 +13,9 @@ public class DataBaseMonitor {
public DataBaseMonitor init() throws SQLException { public DataBaseMonitor init() throws SQLException {
if (conn == null) { if (conn == null) {
String jdbcURL = System.getenv("TDENGINE_JDBC_URL"); String jdbcURL = System.getenv("TDENGINE_JDBC_URL");
if (jdbcURL == null || jdbcURL == ""){
jdbcURL = "jdbc:TAOS://localhost:6030?user=root&password=taosdata";
}
conn = DriverManager.getConnection(jdbcURL); conn = DriverManager.getConnection(jdbcURL);
stmt = conn.createStatement(); stmt = conn.createStatement();
} }

3
docs/examples/java/src/main/java/com/taos/example/highvolume/SQLWriter.java
View File

@ -69,6 +69,9 @@ public class SQLWriter {
*/ */
private static Connection getConnection() throws SQLException { private static Connection getConnection() throws SQLException {
String jdbcURL = System.getenv("TDENGINE_JDBC_URL"); String jdbcURL = System.getenv("TDENGINE_JDBC_URL");
if (jdbcURL == null || jdbcURL == ""){
jdbcURL = "jdbc:TAOS://localhost:6030?user=root&password=taosdata";
}
return DriverManager.getConnection(jdbcURL); return DriverManager.getConnection(jdbcURL);
} }

57
docs/examples/java/src/test/java/com/taos/test/TestAll.java
View File

@ -17,6 +17,37 @@ public class TestAll {
stmt.execute("drop database if exists " + dbName); stmt.execute("drop database if exists " + dbName);
} }
} }
waitTransaction();
}
public void dropTopic(String topicName) throws SQLException {
String jdbcUrl = "jdbc:TAOS://localhost:6030?user=root&password=taosdata";
try (Connection conn = DriverManager.getConnection(jdbcUrl)) {
try (Statement stmt = conn.createStatement()) {
stmt.execute("drop topic if exists " + topicName);
}
}
waitTransaction();
}
public void waitTransaction() throws SQLException {
String jdbcUrl = "jdbc:TAOS://localhost:6030?user=root&password=taosdata";
try (Connection conn = DriverManager.getConnection(jdbcUrl)) {
try (Statement stmt = conn.createStatement()) {
for (int i = 0; i < 10; i++) {
stmt.execute("show transactions");
try (ResultSet resultSet = stmt.getResultSet()) {
if (resultSet.next()) {
int count = resultSet.getInt(1);
if (count == 0) {
break;
}
}
}
}
}
}
} }
public void insertData() throws SQLException { public void insertData() throws SQLException {
@ -104,14 +135,20 @@ public class TestAll {
SubscribeDemo.main(args); SubscribeDemo.main(args);
} }
// @Test @Test
// public void testSubscribeJni() throws SQLException, InterruptedException { public void testSubscribeJni() throws SQLException, InterruptedException {
// dropDB("power"); dropTopic("topic_meters");
// ConsumerLoopFull.main(args); dropDB("power");
// } ConsumerLoopFull.main(args);
// @Test dropTopic("topic_meters");
// public void testSubscribeWs() throws SQLException, InterruptedException { dropDB("power");
// dropDB("power"); }
// WsConsumerLoopFull.main(args); @Test
// } public void testSubscribeWs() throws SQLException, InterruptedException {
dropTopic("topic_meters");
dropDB("power");
WsConsumerLoopFull.main(args);
dropTopic("topic_meters");
dropDB("power");
}
} }

4
docs/zh/04-get-started/01-docker.md
View File

@ -17,7 +17,7 @@ docker pull tdengine/tdengine:latest
或者指定版本的容器镜像： 或者指定版本的容器镜像：
```shell ```shell
docker pull tdengine/tdengine:3.0.1.4 docker pull tdengine/tdengine:3.3.3.0
``` ```
然后只需执行下面的命令： 然后只需执行下面的命令：
@ -121,4 +121,4 @@ SELECT AVG(current), MAX(voltage), MIN(phase) FROM test.meters WHERE groupId = 1
SELECT _wstart, AVG(current), MAX(voltage), MIN(phase) FROM test.d1001 INTERVAL(10s); SELECT _wstart, AVG(current), MAX(voltage), MIN(phase) FROM test.d1001 INTERVAL(10s);
``` ```
在上面的查询中，使用系统提供的伪列_wstart 来给出每个窗口的开始时间。 在上面的查询中，使用系统提供的伪列 _wstart 来给出每个窗口的开始时间。

2
docs/zh/04-get-started/03-package.md
View File

@ -317,4 +317,4 @@ SELECT AVG(current), MAX(voltage), MIN(phase) FROM test.meters WHERE groupId = 1
SELECT _wstart, AVG(current), MAX(voltage), MIN(phase) FROM test.d1001 INTERVAL(10s); SELECT _wstart, AVG(current), MAX(voltage), MIN(phase) FROM test.d1001 INTERVAL(10s);
``` ```
在上面的查询中，使用系统提供的伪列_wstart 来给出每个窗口的开始时间。 在上面的查询中，使用系统提供的伪列 _wstart 来给出每个窗口的开始时间。

2
docs/zh/04-get-started/_07-use.md
View File

@ -54,4 +54,4 @@ SELECT AVG(current), MAX(voltage), MIN(phase) FROM test.meters WHERE groupId = 1
SELECT _wstart, AVG(current), MAX(voltage), MIN(phase) FROM test.d1001 INTERVAL(10s); SELECT _wstart, AVG(current), MAX(voltage), MIN(phase) FROM test.d1001 INTERVAL(10s);
``` ```
在上面的查询中，使用系统提供的伪列_wstart 来给出每个窗口的开始时间。 在上面的查询中，使用系统提供的伪列 _wstart 来给出每个窗口的开始时间。

15
docs/zh/14-reference/01-components/01-taosd.md
View File

@ -163,7 +163,7 @@ charset 的有效值是 UTF-8。
| 参数名称 | 参数说明 | | 参数名称 | 参数说明 |
| :----------------: | :---------------------------------------------: | | :----------------: | :---------------------------------------------: |
| numOfCommitThreads | 写入线程的最大数量，取值范围 0-1024，缺省值为 4 | | numOfCommitThreads | 落盘线程的最大数量，取值范围 0-1024，缺省值为 4 |
### 日志相关 ### 日志相关
@ -223,16 +223,16 @@ lossyColumns float|double
| 参数名称 | 参数说明 | | 参数名称 | 参数说明 |
| :--------------: | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------: | | :--------------: | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------: |
| enableCoreFile | crash 时是否生成 core 文件；0: 不生成，1：生成；默认值 为 1; 不同的启动方式，生成 core 文件的目录如下：1、systemctl start taosd 启动：生成的 core 在根目录下 <br/> 2、手动启动，就在 taosd 执行目录下。 | | enableCoreFile | crash 时是否生成 core 文件；0: 不生成，1：生成；默认值为 1; 不同的启动方式，生成 core 文件的目录如下：1、systemctl start taosd 启动：生成的 core 在根目录下 <br/> 2、手动启动，就在 taosd 执行目录下。 |
| udf | 是否启动 UDF 服务；0: 不启动，1：启动；默认值 为 0 | | udf | 是否启动 UDF 服务；0: 不启动，1：启动；默认值为 0 |
| ttlChangeOnWrite | ttl 到期时间是否伴随表的修改操作改变; 0: 不改变，1：改变 ；默认值 为 | | ttlChangeOnWrite | ttl 到期时间是否伴随表的修改操作改变; 0: 不改变，1：改变；默认值为 0 |
| tmqMaxTopicNum | 订阅最多可建立的 topic 数量; 取值范围 1-10000；缺省值 为20 | | tmqMaxTopicNum | 订阅最多可建立的 topic 数量; 取值范围 1-10000；缺省值为20 |
| maxTsmaNum | 集群内可创建的TSMA个数；取值范围：0-3；缺省值: 3 | | maxTsmaNum | 集群内可创建的TSMA个数；取值范围：0-3；缺省值为 3 |
## taosd 监控指标 ## taosd 监控指标
taosd 会将监控指标上报给 taosKeeper，这些监控指标会被 taosKeeper 写入监控数据库，默认是 `log` 库，可以在 taoskeeper 配置文件中修改。以下是这些监控指标的详细介绍。 taosd 会将监控指标上报给 taosKeeper，这些监控指标会被 taosKeeper 写入监控数据库，默认是 `log` 库，可以在 taoskeeper 配置文件中修改。以下是这些监控指标的详细介绍。
### taosd\_cluster\_basic 表 ### taosd\_cluster\_basic 表
@ -458,4 +458,3 @@ TDengine 的日志文件主要包括普通日志和慢日志两种类型。
3. 多个客户端的日志存储在相应日志路径下的同一个 taosSlowLog.yyyy.mm.dd 文件里。 3. 多个客户端的日志存储在相应日志路径下的同一个 taosSlowLog.yyyy.mm.dd 文件里。
4. 慢日志文件不自动删除，不压缩。 4. 慢日志文件不自动删除，不压缩。
5. 使用和普通日志文件相同的三个参数 logDir, minimalLogDirGB, asyncLog。另外两个参数 numOfLogLines，logKeepDays 不适用于慢日志。 5. 使用和普通日志文件相同的三个参数 logDir, minimalLogDirGB, asyncLog。另外两个参数 numOfLogLines，logKeepDays 不适用于慢日志。

2
docs/zh/14-reference/01-components/07-explorer.md
View File

@ -8,7 +8,7 @@ taosExplorer 是一个为用户提供 TDengine 实例的可视化管理交互工
## 安装 ## 安装
taosEexplorer 无需单独安装，从 TDengine 3.3.0.0 版本开始，它随着 TDengine 安装包一起发布，安装完成后，就可以看到 `taos-explorer` 服务。如果按照 GitHub 里步骤自己编译 TDengine 源代码生成的安装包不包含 taosExplorer。 taosExplorer 无需单独安装，从 TDengine 3.3.0.0 版本开始，它随着 TDengine 安装包一起发布，安装完成后，就可以看到 `taos-explorer` 服务。如果按照 GitHub 里步骤自己编译 TDengine 源代码生成的安装包不包含 taosExplorer。
## 配置 ## 配置

2
docs/zh/14-reference/03-taos-sql/03-table.md
View File

@ -87,7 +87,7 @@ CREATE TABLE [IF NOT EXISTS] USING [db_name.]stb_name (field1_name [, field2_nam
**参数说明** **参数说明**
1. FILE 语法表示数据来自于 CSV 文件（英文逗号分隔、英文单引号括住每个值），CSV 文件无需表头。CSV 文件中应仅包含 table name 与 tag 值。如需插入数据，请参考数据写入章节。 1. FILE 语法表示数据来自于 CSV 文件（英文逗号分隔、英文单引号括住每个值），CSV 文件无需表头。CSV 文件中应仅包含 table name 与 tag 值。如需插入数据，请参考'数据写入'章节。
2. 为指定的 stb_name 创建子表，该超级表必须已经存在。 2. 为指定的 stb_name 创建子表，该超级表必须已经存在。
3. field_name 列表顺序与 CSV 文件各列内容顺序一致。列表中不允许出现重复项，且必须包含 `tbname`，可包含零个或多个超级表中已定义的标签列。未包含在列表中的标签值将被设置为 NULL。 3. field_name 列表顺序与 CSV 文件各列内容顺序一致。列表中不允许出现重复项，且必须包含 `tbname`，可包含零个或多个超级表中已定义的标签列。未包含在列表中的标签值将被设置为 NULL。

26
docs/zh/14-reference/03-taos-sql/05-insert.md
View File

@ -1,7 +1,7 @@
--- ---
sidebar_label: 数据写入 sidebar_label: 数据写入
title: 数据写入 title: 数据写入
description: 写入数据的详细语法 description: 写入数据的详细语法
--- ---
## 写入语法 ## 写入语法
@ -25,9 +25,9 @@ INSERT INTO tb_name [(field1_name, ...)] subquery
### 超级表语法 ### 超级表语法
```sql ```sql
INSERT INTO INSERT INTO
stb1_name [(field1_name, ...)] stb1_name [(field1_name, ...)]
VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
[stb2_name [(field1_name, ...)] [stb2_name [(field1_name, ...)]
VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
...]; ...];
``` ```
@ -47,7 +47,7 @@ INSERT INTO
2. VALUES 语法表示了要插入的一行或多行数据。 2. VALUES 语法表示了要插入的一行或多行数据。
3. FILE 语法表示数据来自于 CSV 文件（英文逗号分隔、英文单引号括住每个值），CSV 文件无需表头。 3. FILE 语法表示数据来自于 CSV 文件（英文逗号分隔、英文单引号括住每个值），CSV 文件无需表头。如仅需创建子表，请参考'表'章节。
4. `INSERT ... VALUES` 语句和 `INSERT ... FILE` 语句均可以在一条 INSERT 语句中同时向多个表插入数据。 4. `INSERT ... VALUES` 语句和 `INSERT ... FILE` 语句均可以在一条 INSERT 语句中同时向多个表插入数据。
@ -154,12 +154,20 @@ INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) FILE '/tmp/c
INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) FILE '/tmp/csvfile_21001.csv' INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) FILE '/tmp/csvfile_21001.csv'
d21002 USING meters (groupId) TAGS (2) FILE '/tmp/csvfile_21002.csv'; d21002 USING meters (groupId) TAGS (2) FILE '/tmp/csvfile_21002.csv';
``` ```
## 超级表语法 ## 向超级表插入数据并自动创建子表
自动建表, 表名通过tbname列指定 自动建表, 表名通过 tbname 列指定
```sql ```sql
INSERT INTO meters(tbname, location, groupId, ts, current, voltage, phase) INSERT INTO meters(tbname, location, groupId, ts, current, voltage, phase)
values('d31001', 'California.SanFrancisco', 2, '2021-07-13 14:06:34.630', 10.2, 219, 0.32) VALUES ('d31001', 'California.SanFrancisco', 2, '2021-07-13 14:06:34.630', 10.2, 219, 0.32)
('d31001', 'California.SanFrancisco', 2, '2021-07-13 14:06:35.779', 10.15, 217, 0.33) ('d31001', 'California.SanFrancisco', 2, '2021-07-13 14:06:35.779', 10.15, 217, 0.33)
('d31002', NULL, 2, '2021-07-13 14:06:34.255', 10.15, 217, 0.33) ('d31002', NULL, 2, '2021-07-13 14:06:34.255', 10.15, 217, 0.33)
```
## 通过 CSV 文件向超级表插入数据并自动创建子表
根据 csv 文件内容，为 超级表创建子表，并填充相应 column 与 tag
```sql
INSERT INTO meters(tbname, location, groupId, ts, current, voltage, phase)
FILE '/tmp/csvfile_21002.csv'
``` ```

6
docs/zh/14-reference/03-taos-sql/21-node.md
View File

@ -27,11 +27,15 @@ SHOW DNODES;
## 删除数据节点 ## 删除数据节点
```sql ```sql
DROP DNODE dnode_id DROP DNODE dnode_id [force] [unsafe]
``` ```
注意删除 dnode 不等于停止相应的进程。实际中推荐先将一个 dnode 删除之后再停止其所对应的进程。 注意删除 dnode 不等于停止相应的进程。实际中推荐先将一个 dnode 删除之后再停止其所对应的进程。
只有在线节点可以被删除。如果要强制删除离线节点，需要执行强制删除操作, 即指定force选项。
当节点上存在单副本，并且节点处于离线，如果要强制删除该节点，需要执行非安全删除，即制定unsafe，并且数据不可再恢复。
## 修改数据节点配置 ## 修改数据节点配置
```sql ```sql

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

@ -151,8 +151,9 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
* @param tversion * @param tversion
* @return * @return
*/ */
int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, int32_t dbNameBuffLen, char* tableName,
int32_t* tversion, int32_t idx, bool* tbGet); int32_t tbaleNameBuffLen, int32_t* sversion, int32_t* tversion, int32_t idx,
bool* tbGet);
/** /**
* The main task execution function, including query on both table and multiple tables, * The main task execution function, including query on both table and multiple tables,

6
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
View File

@ -36,12 +36,13 @@ typedef struct SVnodeMgmt {
SSingleWorker mgmtWorker; SSingleWorker mgmtWorker;
SSingleWorker mgmtMultiWorker; SSingleWorker mgmtMultiWorker;
SHashObj *hash; SHashObj *hash;
SHashObj *closedHash;
TdThreadRwlock lock; TdThreadRwlock lock;
SVnodesStat state; SVnodesStat state;
STfs *pTfs; STfs *pTfs;
TdThread thread; TdThread thread;
bool stop; bool stop;
TdThreadMutex createLock; TdThreadMutex fileLock;
} SVnodeMgmt; } SVnodeMgmt;
typedef struct { typedef struct {
@ -94,7 +95,7 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId);
SVnodeObj *vmAcquireVnodeImpl(SVnodeMgmt *pMgmt, int32_t vgId, bool strict); SVnodeObj *vmAcquireVnodeImpl(SVnodeMgmt *pMgmt, int32_t vgId, bool strict);
void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode); void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode);
int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl); int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl);
void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal); void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal, bool keepClosed);
// vmHandle.c // vmHandle.c
SArray *vmGetMsgHandles(); SArray *vmGetMsgHandles();
@ -111,6 +112,7 @@ int32_t vmProcessArbHeartBeatReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes); int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes);
int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt); int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt);
int32_t vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes); int32_t vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes);
int32_t vmGetAllVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes);
// vmWorker.c // vmWorker.c
int32_t vmStartWorker(SVnodeMgmt *pMgmt); int32_t vmStartWorker(SVnodeMgmt *pMgmt);

85
source/dnode/mgmt/mgmt_vnode/src/vmFile.c
View File

@ -19,6 +19,54 @@
#define MAX_CONTENT_LEN 2 * 1024 * 1024 #define MAX_CONTENT_LEN 2 * 1024 * 1024
int32_t vmGetAllVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes) {
(void)taosThreadRwlockRdlock(&pMgmt->lock);
int32_t num = 0;
int32_t size = taosHashGetSize(pMgmt->hash);
int32_t closedSize = taosHashGetSize(pMgmt->closedHash);
size += closedSize;
SVnodeObj **pVnodes = taosMemoryCalloc(size, sizeof(SVnodeObj *));
if (pVnodes == NULL) {
(void)taosThreadRwlockUnlock(&pMgmt->lock);
return terrno;
}
void *pIter = taosHashIterate(pMgmt->hash, NULL);
while (pIter) {
SVnodeObj **ppVnode = pIter;
SVnodeObj *pVnode = *ppVnode;
if (pVnode && num < size) {
int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
// dTrace("vgId:%d, acquire vnode list, ref:%d", pVnode->vgId, refCount);
pVnodes[num++] = (*ppVnode);
pIter = taosHashIterate(pMgmt->hash, pIter);
} else {
taosHashCancelIterate(pMgmt->hash, pIter);
}
}
pIter = taosHashIterate(pMgmt->closedHash, NULL);
while (pIter) {
SVnodeObj **ppVnode = pIter;
SVnodeObj *pVnode = *ppVnode;
if (pVnode && num < size) {
int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
// dTrace("vgId:%d, acquire vnode list, ref:%d", pVnode->vgId, refCount);
pVnodes[num++] = (*ppVnode);
pIter = taosHashIterate(pMgmt->closedHash, pIter);
} else {
taosHashCancelIterate(pMgmt->closedHash, pIter);
}
}
(void)taosThreadRwlockUnlock(&pMgmt->lock);
*numOfVnodes = num;
*ppVnodes = pVnodes;
return 0;
}
int32_t vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes) { int32_t vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes) {
(void)taosThreadRwlockRdlock(&pMgmt->lock); (void)taosThreadRwlockRdlock(&pMgmt->lock);
@ -203,6 +251,8 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
SVnodeObj **ppVnodes = NULL; SVnodeObj **ppVnodes = NULL;
char file[PATH_MAX] = {0}; char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0}; char realfile[PATH_MAX] = {0};
int32_t lino = 0;
int32_t ret = -1;
int32_t nBytes = snprintf(file, sizeof(file), "%s%svnodes_tmp.json", pMgmt->path, TD_DIRSEP); int32_t nBytes = snprintf(file, sizeof(file), "%s%svnodes_tmp.json", pMgmt->path, TD_DIRSEP);
if (nBytes <= 0 || nBytes >= sizeof(file)) { if (nBytes <= 0 || nBytes >= sizeof(file)) {
@ -215,8 +265,7 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
} }
int32_t numOfVnodes = 0; int32_t numOfVnodes = 0;
code = vmGetVnodeListFromHash(pMgmt, &numOfVnodes, &ppVnodes); TAOS_CHECK_GOTO(vmGetAllVnodeListFromHash(pMgmt, &numOfVnodes, &ppVnodes), &lino, _OVER);
if (code) goto _OVER;
// terrno = TSDB_CODE_OUT_OF_MEMORY; // terrno = TSDB_CODE_OUT_OF_MEMORY;
pJson = tjsonCreateObject(); pJson = tjsonCreateObject();
@ -224,39 +273,56 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
code = terrno; code = terrno;
goto _OVER; goto _OVER;
} }
if ((code = vmEncodeVnodeList(pJson, ppVnodes, numOfVnodes)) != 0) goto _OVER; TAOS_CHECK_GOTO(vmEncodeVnodeList(pJson, ppVnodes, numOfVnodes), &lino, _OVER);
buffer = tjsonToString(pJson); buffer = tjsonToString(pJson);
if (buffer == NULL) { if (buffer == NULL) {
code = TSDB_CODE_INVALID_JSON_FORMAT; code = TSDB_CODE_INVALID_JSON_FORMAT;
lino = __LINE__;
goto _OVER;
}
code = taosThreadMutexLock(&pMgmt->fileLock);
if (code != 0) {
lino = __LINE__;
goto _OVER; goto _OVER;
} }
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH); pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) { if (pFile == NULL) {
code = terrno; code = terrno;
goto _OVER; lino = __LINE__;
goto _OVER1;
} }
int32_t len = strlen(buffer); int32_t len = strlen(buffer);
if (taosWriteFile(pFile, buffer, len) <= 0) { if (taosWriteFile(pFile, buffer, len) <= 0) {
code = terrno; code = terrno;
goto _OVER; lino = __LINE__;
goto _OVER1;
} }
if (taosFsyncFile(pFile) < 0) { if (taosFsyncFile(pFile) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _OVER; lino = __LINE__;
goto _OVER1;
} }
code = taosCloseFile(&pFile); code = taosCloseFile(&pFile);
if (code != 0) { if (code != 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _OVER; lino = __LINE__;
goto _OVER1;
} }
TAOS_CHECK_GOTO(taosRenameFile(file, realfile), NULL, _OVER); TAOS_CHECK_GOTO(taosRenameFile(file, realfile), &lino, _OVER1);
dInfo("succeed to write vnodes file:%s, vnodes:%d", realfile, numOfVnodes); dInfo("succeed to write vnodes file:%s, vnodes:%d", realfile, numOfVnodes);
_OVER1:
ret = taosThreadMutexUnlock(&pMgmt->fileLock);
if (ret != 0) {
dError("failed to unlock since %s", tstrerror(ret));
}
_OVER: _OVER:
if (pJson != NULL) tjsonDelete(pJson); if (pJson != NULL) tjsonDelete(pJson);
if (buffer != NULL) taosMemoryFree(buffer); if (buffer != NULL) taosMemoryFree(buffer);
@ -272,7 +338,8 @@ _OVER:
} }
if (code != 0) { if (code != 0) {
dError("failed to write vnodes file:%s since %s, vnodes:%d", realfile, tstrerror(code), numOfVnodes); dError("failed to write vnodes file:%s at line:%d since %s, vnodes:%d", realfile, lino, tstrerror(code),
numOfVnodes);
} }
return code; return code;
} }

37
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -415,27 +415,30 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
goto _OVER; goto _OVER;
} }
code = taosThreadMutexLock(&pMgmt->createLock);
if (code != 0) {
dError("vgId:%d, failed to lock since %s", req.vgId, tstrerror(code));
goto _OVER;
}
code = vmWriteVnodeListToFile(pMgmt); code = vmWriteVnodeListToFile(pMgmt);
if (code != 0) { if (code != 0) {
code = terrno != 0 ? terrno : code; code = terrno != 0 ? terrno : code;
int32_t ret = taosThreadMutexUnlock(&pMgmt->createLock);
if (ret != 0) {
dError("vgId:%d, failed to unlock since %s", req.vgId, tstrerror(ret));
}
goto _OVER; goto _OVER;
} }
int32_t ret = taosThreadMutexUnlock(&pMgmt->createLock);
if (ret != 0) {
dError("vgId:%d, failed to unlock since %s", req.vgId, tstrerror(ret));
}
_OVER: _OVER:
if (code != 0) { if (code != 0) {
int32_t r = 0;
r = taosThreadRwlockWrlock(&pMgmt->lock);
if (r != 0) {
dError("vgId:%d, failed to lock since %s", req.vgId, tstrerror(r));
}
if (r == 0) {
dInfo("vgId:%d, remove from hash", req.vgId);
r = taosHashRemove(pMgmt->hash, &req.vgId, sizeof(int32_t));
if (r != 0) {
dError("vgId:%d, failed to remove vnode since %s", req.vgId, tstrerror(r));
}
}
r = taosThreadRwlockUnlock(&pMgmt->lock);
if (r != 0) {
dError("vgId:%d, failed to unlock since %s", req.vgId, tstrerror(r));
}
vnodeClose(pImpl); vnodeClose(pImpl);
vnodeDestroy(0, path, pMgmt->pTfs, 0); vnodeDestroy(0, path, pMgmt->pTfs, 0);
} else { } else {
@ -535,7 +538,7 @@ int32_t vmProcessAlterVnodeTypeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path)); tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
bool commitAndRemoveWal = vnodeShouldRemoveWal(pVnode->pImpl); bool commitAndRemoveWal = vnodeShouldRemoveWal(pVnode->pImpl);
vmCloseVnode(pMgmt, pVnode, commitAndRemoveWal); vmCloseVnode(pMgmt, pVnode, commitAndRemoveWal, true);
int32_t diskPrimary = wrapperCfg.diskPrimary; int32_t diskPrimary = wrapperCfg.diskPrimary;
char path[TSDB_FILENAME_LEN] = {0}; char path[TSDB_FILENAME_LEN] = {0};
@ -683,7 +686,7 @@ int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
} }
dInfo("vgId:%d, close vnode", srcVgId); dInfo("vgId:%d, close vnode", srcVgId);
vmCloseVnode(pMgmt, pVnode, true); vmCloseVnode(pMgmt, pVnode, true, false);
int32_t diskPrimary = wrapperCfg.diskPrimary; int32_t diskPrimary = wrapperCfg.diskPrimary;
char srcPath[TSDB_FILENAME_LEN] = {0}; char srcPath[TSDB_FILENAME_LEN] = {0};
@ -792,7 +795,7 @@ int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path)); tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
bool commitAndRemoveWal = vnodeShouldRemoveWal(pVnode->pImpl); bool commitAndRemoveWal = vnodeShouldRemoveWal(pVnode->pImpl);
vmCloseVnode(pMgmt, pVnode, commitAndRemoveWal); vmCloseVnode(pMgmt, pVnode, commitAndRemoveWal, true);
int32_t diskPrimary = wrapperCfg.diskPrimary; int32_t diskPrimary = wrapperCfg.diskPrimary;
char path[TSDB_FILENAME_LEN] = {0}; char path[TSDB_FILENAME_LEN] = {0};
@ -860,7 +863,7 @@ int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return code; return code;
} }
vmCloseVnode(pMgmt, pVnode, false); vmCloseVnode(pMgmt, pVnode, false, false);
if (vmWriteVnodeListToFile(pMgmt) != 0) { if (vmWriteVnodeListToFile(pMgmt) != 0) {
dError("vgId:%d, failed to write vnode list since %s", vgId, terrstr()); dError("vgId:%d, failed to write vnode list since %s", vgId, terrstr());
} }

81
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
View File

@ -166,16 +166,34 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
(void)taosThreadRwlockWrlock(&pMgmt->lock); (void)taosThreadRwlockWrlock(&pMgmt->lock);
SVnodeObj *pOld = NULL; SVnodeObj *pOld = NULL;
int32_t r = taosHashGetDup(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld); int32_t r = taosHashGetDup(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld);
if (r != 0) {
dError("vgId:%d, failed to get vnode from hash", pVnode->vgId);
}
if (pOld) { if (pOld) {
vmFreeVnodeObj(&pOld); vmFreeVnodeObj(&pOld);
} }
int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *)); int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *));
pOld = NULL;
r = taosHashGetDup(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld);
if (r != 0) {
dError("vgId:%d, failed to get vnode from closedHash", pVnode->vgId);
}
if (pOld) {
vmFreeVnodeObj(&pOld);
}
dInfo("vgId:%d, remove from closedHash", pVnode->vgId);
r = taosHashRemove(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t));
if (r != 0) {
dError("vgId:%d, failed to remove vnode from hash", pVnode->vgId);
}
(void)taosThreadRwlockUnlock(&pMgmt->lock); (void)taosThreadRwlockUnlock(&pMgmt->lock);
return code; return code;
} }
void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal) { void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal, bool keepClosed) {
char path[TSDB_FILENAME_LEN] = {0}; char path[TSDB_FILENAME_LEN] = {0};
bool atExit = true; bool atExit = true;
@ -185,7 +203,40 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
(void)taosThreadRwlockWrlock(&pMgmt->lock); (void)taosThreadRwlockWrlock(&pMgmt->lock);
int32_t r = taosHashRemove(pMgmt->hash, &pVnode->vgId, sizeof(int32_t)); int32_t r = taosHashRemove(pMgmt->hash, &pVnode->vgId, sizeof(int32_t));
if (r != 0) {
dError("vgId:%d, failed to remove vnode from hash", pVnode->vgId);
}
if (keepClosed) {
SVnodeObj *pClosedVnode = taosMemoryCalloc(1, sizeof(SVnodeObj));
(void)memset(pClosedVnode, 0, sizeof(SVnodeObj));
if (pVnode == NULL) {
dError("vgId:%d, failed to alloc vnode since %s", pVnode->vgId, terrstr());
(void)taosThreadRwlockUnlock(&pMgmt->lock);
return;
}
pClosedVnode->vgId = pVnode->vgId;
pClosedVnode->dropped = pVnode->dropped;
pClosedVnode->vgVersion = pVnode->vgVersion;
pClosedVnode->diskPrimary = pVnode->diskPrimary;
pClosedVnode->toVgId = pVnode->toVgId;
SVnodeObj *pOld = NULL;
r = taosHashGetDup(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t), (void *)&pOld);
if (r != 0) {
dError("vgId:%d, failed to get vnode from closedHash", pVnode->vgId);
}
if (pOld) {
vmFreeVnodeObj(&pOld);
}
dInfo("vgId:%d, put vnode to closedHash", pVnode->vgId);
r = taosHashPut(pMgmt->closedHash, &pVnode->vgId, sizeof(int32_t), &pClosedVnode, sizeof(SVnodeObj *));
if (r != 0) {
dError("vgId:%d, failed to put vnode to closedHash", pVnode->vgId);
}
}
(void)taosThreadRwlockUnlock(&pMgmt->lock); (void)taosThreadRwlockUnlock(&pMgmt->lock);
vmReleaseVnode(pMgmt, pVnode); vmReleaseVnode(pMgmt, pVnode);
if (pVnode->failed) { if (pVnode->failed) {
@ -362,9 +413,15 @@ static void *vmOpenVnodeInThread(void *param) {
static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) { static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) {
pMgmt->hash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK); pMgmt->hash = taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
if (pMgmt->hash == NULL) { if (pMgmt->hash == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
dError("failed to init vnode hash since %s", terrstr()); dError("failed to init vnode hash since %s", terrstr());
return -1; return TSDB_CODE_OUT_OF_MEMORY;
}
pMgmt->closedHash =
taosHashInit(TSDB_MIN_VNODES, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
if (pMgmt->hash == NULL) {
dError("failed to init vnode closed hash since %s", terrstr());
return TSDB_CODE_OUT_OF_MEMORY;
} }
SWrapperCfg *pCfgs = NULL; SWrapperCfg *pCfgs = NULL;
@ -459,7 +516,7 @@ static void *vmCloseVnodeInThread(void *param) {
pMgmt->state.openVnodes, pMgmt->state.totalVnodes); pMgmt->state.openVnodes, pMgmt->state.totalVnodes);
tmsgReportStartup("vnode-close", stepDesc); tmsgReportStartup("vnode-close", stepDesc);
vmCloseVnode(pMgmt, pVnode, false); vmCloseVnode(pMgmt, pVnode, false, false);
} }
dInfo("thread:%d, numOfVnodes:%d is closed", pThread->threadIndex, pThread->vnodeNum); dInfo("thread:%d, numOfVnodes:%d is closed", pThread->threadIndex, pThread->vnodeNum);
@ -537,6 +594,18 @@ static void vmCloseVnodes(SVnodeMgmt *pMgmt) {
pMgmt->hash = NULL; pMgmt->hash = NULL;
} }
void *pIter = taosHashIterate(pMgmt->closedHash, NULL);
while (pIter) {
SVnodeObj **ppVnode = pIter;
vmFreeVnodeObj(ppVnode);
pIter = taosHashIterate(pMgmt->closedHash, pIter);
}
if (pMgmt->closedHash != NULL) {
taosHashCleanup(pMgmt->closedHash);
pMgmt->closedHash = NULL;
}
dInfo("total vnodes:%d are all closed", numOfVnodes); dInfo("total vnodes:%d are all closed", numOfVnodes);
} }
@ -545,7 +614,7 @@ static void vmCleanup(SVnodeMgmt *pMgmt) {
vmStopWorker(pMgmt); vmStopWorker(pMgmt);
vnodeCleanup(); vnodeCleanup();
(void)taosThreadRwlockDestroy(&pMgmt->lock); (void)taosThreadRwlockDestroy(&pMgmt->lock);
(void)taosThreadMutexDestroy(&pMgmt->createLock); (void)taosThreadMutexDestroy(&pMgmt->fileLock);
taosMemoryFree(pMgmt); taosMemoryFree(pMgmt);
} }
@ -637,7 +706,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
goto _OVER; goto _OVER;
} }
code = taosThreadMutexInit(&pMgmt->createLock, NULL); code = taosThreadMutexInit(&pMgmt->fileLock, NULL);
if (code != 0) { if (code != 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _OVER; goto _OVER;

15
source/dnode/mnode/impl/src/mndArbGroup.c
View File

@ -15,13 +15,10 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "mndArbGroup.h" #include "mndArbGroup.h"
#include "audit.h"
#include "mndDb.h" #include "mndDb.h"
#include "mndDnode.h" #include "mndDnode.h"
#include "mndPrivilege.h"
#include "mndShow.h" #include "mndShow.h"
#include "mndTrans.h" #include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h" #include "mndVgroup.h"
#define ARBGROUP_VER_NUMBER 1 #define ARBGROUP_VER_NUMBER 1
@ -245,11 +242,11 @@ static int32_t mndArbGroupActionUpdate(SSdb *pSdb, SArbGroup *pOld, SArbGroup *p
} }
for (int i = 0; i < TSDB_ARB_GROUP_MEMBER_NUM; i++) { for (int i = 0; i < TSDB_ARB_GROUP_MEMBER_NUM; i++) {
(void)memcpy(pOld->members[i].state.token, pNew->members[i].state.token, TSDB_ARB_TOKEN_SIZE); tstrncpy(pOld->members[i].state.token, pNew->members[i].state.token, TSDB_ARB_TOKEN_SIZE);
} }
pOld->isSync = pNew->isSync; pOld->isSync = pNew->isSync;
pOld->assignedLeader.dnodeId = pNew->assignedLeader.dnodeId; pOld->assignedLeader.dnodeId = pNew->assignedLeader.dnodeId;
(void)memcpy(pOld->assignedLeader.token, pNew->assignedLeader.token, TSDB_ARB_TOKEN_SIZE); tstrncpy(pOld->assignedLeader.token, pNew->assignedLeader.token, TSDB_ARB_TOKEN_SIZE);
pOld->assignedLeader.acked = pNew->assignedLeader.acked; pOld->assignedLeader.acked = pNew->assignedLeader.acked;
pOld->version++; pOld->version++;
@ -834,12 +831,12 @@ static int32_t mndProcessArbUpdateGroupBatchReq(SRpcMsg *pReq) {
newGroup.dbUid = pUpdateGroup->dbUid; newGroup.dbUid = pUpdateGroup->dbUid;
for (int i = 0; i < TSDB_ARB_GROUP_MEMBER_NUM; i++) { for (int i = 0; i < TSDB_ARB_GROUP_MEMBER_NUM; i++) {
newGroup.members[i].info.dnodeId = pUpdateGroup->members[i].dnodeId; newGroup.members[i].info.dnodeId = pUpdateGroup->members[i].dnodeId;
(void)memcpy(newGroup.members[i].state.token, pUpdateGroup->members[i].token, TSDB_ARB_TOKEN_SIZE); tstrncpy(newGroup.members[i].state.token, pUpdateGroup->members[i].token, TSDB_ARB_TOKEN_SIZE);
} }
newGroup.isSync = pUpdateGroup->isSync; newGroup.isSync = pUpdateGroup->isSync;
newGroup.assignedLeader.dnodeId = pUpdateGroup->assignedLeader.dnodeId; newGroup.assignedLeader.dnodeId = pUpdateGroup->assignedLeader.dnodeId;
(void)memcpy(newGroup.assignedLeader.token, pUpdateGroup->assignedLeader.token, TSDB_ARB_TOKEN_SIZE); tstrncpy(newGroup.assignedLeader.token, pUpdateGroup->assignedLeader.token, TSDB_ARB_TOKEN_SIZE);
newGroup.assignedLeader.acked = pUpdateGroup->assignedLeader.acked; newGroup.assignedLeader.acked = pUpdateGroup->assignedLeader.acked;
newGroup.version = pUpdateGroup->version; newGroup.version = pUpdateGroup->version;
@ -897,7 +894,7 @@ static void mndArbGroupSetAssignedLeader(SArbGroup *pGroup, int32_t index) {
SArbGroupMember *pMember = &pGroup->members[index]; SArbGroupMember *pMember = &pGroup->members[index];
pGroup->assignedLeader.dnodeId = pMember->info.dnodeId; pGroup->assignedLeader.dnodeId = pMember->info.dnodeId;
(void)strncpy(pGroup->assignedLeader.token, pMember->state.token, TSDB_ARB_TOKEN_SIZE); tstrncpy(pGroup->assignedLeader.token, pMember->state.token, TSDB_ARB_TOKEN_SIZE);
pGroup->assignedLeader.acked = false; pGroup->assignedLeader.acked = false;
} }
@ -979,7 +976,7 @@ bool mndUpdateArbGroupByHeartBeat(SArbGroup *pGroup, SVArbHbRspMember *pRspMembe
// update token // update token
mndArbGroupDupObj(pGroup, pNewGroup); mndArbGroupDupObj(pGroup, pNewGroup);
(void)memcpy(pNewGroup->members[index].state.token, pRspMember->memberToken, TSDB_ARB_TOKEN_SIZE); tstrncpy(pNewGroup->members[index].state.token, pRspMember->memberToken, TSDB_ARB_TOKEN_SIZE);
pNewGroup->isSync = false; pNewGroup->isSync = false;
bool resetAssigned = false; bool resetAssigned = false;

11
source/dnode/mnode/sdb/src/sdbFile.c
View File

@ -400,8 +400,8 @@ static int32_t sdbReadFileImp(SSdb *pSdb) {
pSdb->commitTerm = pSdb->applyTerm; pSdb->commitTerm = pSdb->applyTerm;
pSdb->commitConfig = pSdb->applyConfig; pSdb->commitConfig = pSdb->applyConfig;
memcpy(pSdb->tableVer, tableVer, sizeof(tableVer)); memcpy(pSdb->tableVer, tableVer, sizeof(tableVer));
mInfo("read sdb file:%s success, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64, file, pSdb->commitIndex, mInfo("vgId:1, trans:0, read sdb file:%s success, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64, file,
pSdb->commitTerm, pSdb->commitConfig); pSdb->commitIndex, pSdb->commitTerm, pSdb->commitConfig);
_OVER: _OVER:
if ((ret = taosCloseFile(&pFile)) != 0) { if ((ret = taosCloseFile(&pFile)) != 0) {
@ -573,7 +573,8 @@ static int32_t sdbWriteFileImp(SSdb *pSdb, int32_t skip_type) {
pSdb->commitIndex = pSdb->applyIndex; pSdb->commitIndex = pSdb->applyIndex;
pSdb->commitTerm = pSdb->applyTerm; pSdb->commitTerm = pSdb->applyTerm;
pSdb->commitConfig = pSdb->applyConfig; pSdb->commitConfig = pSdb->applyConfig;
mInfo("write sdb file success, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " file:%s", mInfo("vgId:1, trans:0, write sdb file success, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64
" file:%s",
pSdb->commitIndex, pSdb->commitTerm, pSdb->commitConfig, curfile); pSdb->commitIndex, pSdb->commitTerm, pSdb->commitConfig, curfile);
} }
@ -610,8 +611,8 @@ int32_t sdbWriteFile(SSdb *pSdb, int32_t delta) {
if (code != 0) { if (code != 0) {
mError("failed to write sdb file since %s", tstrerror(code)); mError("failed to write sdb file since %s", tstrerror(code));
} else { } else {
mInfo("write sdb file success, apply index:%" PRId64 " term:%" PRId64 " config:%" PRId64, pSdb->applyIndex, mInfo("vgId:1, trans:0, write sdb file success, apply index:%" PRId64 ", term:%" PRId64 ", config:%" PRId64,
pSdb->applyTerm, pSdb->applyConfig); pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig);
} }
(void)taosThreadMutexUnlock(&pSdb->filelock); (void)taosThreadMutexUnlock(&pSdb->filelock);
return code; return code;

91
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -723,34 +723,32 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
{ {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol); if (values_list[0] != NULL) {
if (code == TSDB_CODE_INVALID_PARA) { code = tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol);
tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, if (code != TSDB_CODE_SUCCESS) {
tstrerror(code)); tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
} else if (code != TSDB_CODE_SUCCESS) { tstrerror(code));
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, goto _exit;
tstrerror(code)); }
goto _exit; if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[0], klen);
}
taosMemoryFreeClear(pLastCol);
} }
if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[0], klen);
}
taosMemoryFreeClear(pLastCol);
pLastCol = NULL; pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol); if (values_list[1] != NULL) {
if (code == TSDB_CODE_INVALID_PARA) { code = tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol);
tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, if (code != TSDB_CODE_SUCCESS) {
tstrerror(code)); tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
} else if (code != TSDB_CODE_SUCCESS) { tstrerror(code));
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, goto _exit;
tstrerror(code)); }
goto _exit; if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[1], klen);
}
taosMemoryFreeClear(pLastCol);
} }
if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[1], klen);
}
taosMemoryFreeClear(pLastCol);
rocksdb_free(values_list[0]); rocksdb_free(values_list[0]);
rocksdb_free(values_list[1]); rocksdb_free(values_list[1]);
@ -1218,14 +1216,13 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
SColVal *pColVal = &updCtx->colVal; SColVal *pColVal = &updCtx->colVal;
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); if (values_list[i] != NULL) {
if (code == TSDB_CODE_INVALID_PARA) { code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, if (code != TSDB_CODE_SUCCESS) {
tstrerror(code)); tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
} else if (code != TSDB_CODE_SUCCESS) { tstrerror(code));
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, goto _exit;
tstrerror(code)); }
goto _exit;
} }
/* /*
if (code) { if (code) {
@ -1692,14 +1689,13 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
continue; continue;
} }
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); if (values_list[i] != NULL) {
if (code == TSDB_CODE_INVALID_PARA) { code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, if (code != TSDB_CODE_SUCCESS) {
tstrerror(code)); tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
} else if (code != TSDB_CODE_SUCCESS) { tstrerror(code));
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, goto _exit;
tstrerror(code)); }
goto _exit;
} }
SLastCol *pToFree = pLastCol; SLastCol *pToFree = pLastCol;
SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j]; SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j];
@ -1959,14 +1955,13 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
for (int i = 0; i < numKeys; ++i) { for (int i = 0; i < numKeys; ++i) {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); if (values_list[i] != NULL) {
if (code == TSDB_CODE_INVALID_PARA) { code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, if (code != TSDB_CODE_SUCCESS) {
tstrerror(code)); tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
} else if (code != TSDB_CODE_SUCCESS) { tstrerror(code));
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, goto _exit;
tstrerror(code)); }
goto _exit;
} }
SIdxKey *idxKey = taosArrayGet(remainCols, i); SIdxKey *idxKey = taosArrayGet(remainCols, i);
SLastKey *pLastKey = &idxKey->key; SLastKey *pLastKey = &idxKey->key;

2
source/libs/executor/src/anomalywindowoperator.c
View File

@ -86,7 +86,7 @@ int32_t createAnomalywindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* p
pOperator->exprSupp.hasWindowOrGroup = true; pOperator->exprSupp.hasWindowOrGroup = true;
pInfo->tsSlotId = ((SColumnNode*)pAnomalyNode->window.pTspk)->slotId; pInfo->tsSlotId = ((SColumnNode*)pAnomalyNode->window.pTspk)->slotId;
strncpy(pInfo->anomalyOpt, pAnomalyNode->anomalyOpt, sizeof(pInfo->anomalyOpt)); tstrncpy(pInfo->anomalyOpt, pAnomalyNode->anomalyOpt, sizeof(pInfo->anomalyOpt));
if (pAnomalyNode->window.pExprs != NULL) { if (pAnomalyNode->window.pExprs != NULL) {
int32_t numOfScalarExpr = 0; int32_t numOfScalarExpr = 0;

2
source/libs/executor/src/exchangeoperator.c
View File

@ -320,7 +320,7 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const
if (!pInfo->pTaskId) { if (!pInfo->pTaskId) {
return terrno; return terrno;
} }
strncpy(pInfo->pTaskId, id, len); tstrncpy(pInfo->pTaskId, id, len);
for (int32_t i = 0; i < numOfSources; ++i) { for (int32_t i = 0; i < numOfSources; ++i) {
SSourceDataInfo dataInfo = {0}; SSourceDataInfo dataInfo = {0};
dataInfo.status = EX_SOURCE_DATA_NOT_READY; dataInfo.status = EX_SOURCE_DATA_NOT_READY;

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

@ -545,8 +545,9 @@ int32_t qUpdateTableListForStreamScanner(qTaskInfo_t tinfo, const SArray* tableI
return code; return code;
} }
int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, int32_t dbNameBuffLen, char* tableName,
int32_t* tversion, int32_t idx, bool* tbGet) { int32_t tbaleNameBuffLen, int32_t* sversion, int32_t* tversion, int32_t idx,
bool* tbGet) {
*tbGet = false; *tbGet = false;
if (tinfo == NULL || dbName == NULL || tableName == NULL) { if (tinfo == NULL || dbName == NULL || tableName == NULL) {
@ -567,12 +568,12 @@ int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* table
*sversion = pSchemaInfo->sw->version; *sversion = pSchemaInfo->sw->version;
*tversion = pSchemaInfo->tversion; *tversion = pSchemaInfo->tversion;
if (pSchemaInfo->dbname) { if (pSchemaInfo->dbname) {
strcpy(dbName, pSchemaInfo->dbname); tstrncpy(dbName, pSchemaInfo->dbname, dbNameBuffLen);
} else { } else {
dbName[0] = 0; dbName[0] = 0;
} }
if (pSchemaInfo->tablename) { if (pSchemaInfo->tablename) {
strcpy(tableName, pSchemaInfo->tablename); tstrncpy(tableName, pSchemaInfo->tablename, tbaleNameBuffLen);
} else { } else {
tableName[0] = 0; tableName[0] = 0;
} }

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

@ -6685,7 +6685,7 @@ int32_t fillTableCountScanDataBlock(STableCountScanSupp* pSupp, char* dbName, ch
QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno); QUERY_CHECK_NULL(colInfoData, code, lino, _end, terrno);
if (strlen(stbName) != 0) { if (strlen(stbName) != 0) {
char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; char varStbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
strncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN); tstrncpy(varDataVal(varStbName), stbName, TSDB_TABLE_NAME_LEN);
varDataSetLen(varStbName, strlen(stbName)); varDataSetLen(varStbName, strlen(stbName));
code = colDataSetVal(colInfoData, 0, varStbName, false); code = colDataSetVal(colInfoData, 0, varStbName, false);
QUERY_CHECK_CODE(code, lino, _end); QUERY_CHECK_CODE(code, lino, _end);

64
source/libs/executor/src/sysscanoperator.c
View File

@ -425,7 +425,7 @@ static bool sysTableIsOperatorCondOnOneTable(SNode* pCond, char* condTable) {
SValueNode* pValue = (SValueNode*)node->pRight; SValueNode* pValue = (SValueNode*)node->pRight;
if (pValue->node.resType.type == TSDB_DATA_TYPE_NCHAR || pValue->node.resType.type == TSDB_DATA_TYPE_VARCHAR) { if (pValue->node.resType.type == TSDB_DATA_TYPE_NCHAR || pValue->node.resType.type == TSDB_DATA_TYPE_VARCHAR) {
char* value = nodesGetValueFromNode(pValue); char* value = nodesGetValueFromNode(pValue);
strncpy(condTable, varDataVal(value), TSDB_TABLE_NAME_LEN); tstrncpy(condTable, varDataVal(value), TSDB_TABLE_NAME_LEN);
return true; return true;
} }
} }
@ -914,41 +914,41 @@ _end:
} }
} }
int32_t convertTagDataToStr(char* str, int type, void* buf, int32_t bufSize, int32_t* len) { int32_t convertTagDataToStr(char* str, int32_t strBuffLen, int type, void* buf, int32_t bufSize, int32_t* len) {
int32_t n = 0; int32_t n = 0;
switch (type) { switch (type) {
case TSDB_DATA_TYPE_NULL: case TSDB_DATA_TYPE_NULL:
n = sprintf(str, "null"); n = tsnprintf(str, strBuffLen, "null");
break; break;
case TSDB_DATA_TYPE_BOOL: case TSDB_DATA_TYPE_BOOL:
n = sprintf(str, (*(int8_t*)buf) ? "true" : "false"); n = tsnprintf(str, strBuffLen, (*(int8_t*)buf) ? "true" : "false");
break; break;
case TSDB_DATA_TYPE_TINYINT: case TSDB_DATA_TYPE_TINYINT:
n = sprintf(str, "%d", *(int8_t*)buf); n = tsnprintf(str, strBuffLen, "%d", *(int8_t*)buf);
break; break;
case TSDB_DATA_TYPE_SMALLINT: case TSDB_DATA_TYPE_SMALLINT:
n = sprintf(str, "%d", *(int16_t*)buf); n = tsnprintf(str, strBuffLen, "%d", *(int16_t*)buf);
break; break;
case TSDB_DATA_TYPE_INT: case TSDB_DATA_TYPE_INT:
n = sprintf(str, "%d", *(int32_t*)buf); n = tsnprintf(str, strBuffLen, "%d", *(int32_t*)buf);
break; break;
case TSDB_DATA_TYPE_BIGINT: case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
n = sprintf(str, "%" PRId64, *(int64_t*)buf); n = tsnprintf(str, strBuffLen, "%" PRId64, *(int64_t*)buf);
break; break;
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
n = sprintf(str, "%.5f", GET_FLOAT_VAL(buf)); n = tsnprintf(str, strBuffLen, "%.5f", GET_FLOAT_VAL(buf));
break; break;
case TSDB_DATA_TYPE_DOUBLE: case TSDB_DATA_TYPE_DOUBLE:
n = sprintf(str, "%.9f", GET_DOUBLE_VAL(buf)); n = tsnprintf(str, strBuffLen, "%.9f", GET_DOUBLE_VAL(buf));
break; break;
case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_BINARY:
@ -973,19 +973,19 @@ int32_t convertTagDataToStr(char* str, int type, void* buf, int32_t bufSize, int
n = length; n = length;
break; break;
case TSDB_DATA_TYPE_UTINYINT: case TSDB_DATA_TYPE_UTINYINT:
n = sprintf(str, "%u", *(uint8_t*)buf); n = tsnprintf(str, strBuffLen, "%u", *(uint8_t*)buf);
break; break;
case TSDB_DATA_TYPE_USMALLINT: case TSDB_DATA_TYPE_USMALLINT:
n = sprintf(str, "%u", *(uint16_t*)buf); n = tsnprintf(str, strBuffLen, "%u", *(uint16_t*)buf);
break; break;
case TSDB_DATA_TYPE_UINT: case TSDB_DATA_TYPE_UINT:
n = sprintf(str, "%u", *(uint32_t*)buf); n = tsnprintf(str, strBuffLen, "%u", *(uint32_t*)buf);
break; break;
case TSDB_DATA_TYPE_UBIGINT: case TSDB_DATA_TYPE_UBIGINT:
n = sprintf(str, "%" PRIu64, *(uint64_t*)buf); n = tsnprintf(str, strBuffLen, "%" PRIu64, *(uint64_t*)buf);
break; break;
default: default:
@ -1065,14 +1065,21 @@ static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo,
int8_t tagType = (*smrSuperTable).me.stbEntry.schemaTag.pSchema[i].type; int8_t tagType = (*smrSuperTable).me.stbEntry.schemaTag.pSchema[i].type;
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 4); pColInfoData = taosArrayGet(dataBlock->pDataBlock, 4);
QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno); QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno);
int32_t tagStrBufflen = 32;
char tagTypeStr[VARSTR_HEADER_SIZE + 32]; char tagTypeStr[VARSTR_HEADER_SIZE + 32];
int tagTypeLen = sprintf(varDataVal(tagTypeStr), "%s", tDataTypes[tagType].name); int tagTypeLen = tsnprintf(varDataVal(tagTypeStr), tagStrBufflen, "%s", tDataTypes[tagType].name);
tagStrBufflen -= tagTypeLen;
if (tagStrBufflen <= 0) {
code = TSDB_CODE_INVALID_PARA;
QUERY_CHECK_CODE(code, lino, _end);
}
if (tagType == TSDB_DATA_TYPE_NCHAR) { if (tagType == TSDB_DATA_TYPE_NCHAR) {
tagTypeLen += sprintf( tagTypeLen += tsnprintf(
varDataVal(tagTypeStr) + tagTypeLen, "(%d)", varDataVal(tagTypeStr) + tagTypeLen, tagStrBufflen, "(%d)",
(int32_t)(((*smrSuperTable).me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)); (int32_t)(((*smrSuperTable).me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
} else if (IS_VAR_DATA_TYPE(tagType)) { } else if (IS_VAR_DATA_TYPE(tagType)) {
tagTypeLen += sprintf(varDataVal(tagTypeStr) + tagTypeLen, "(%d)", tagTypeLen += tsnprintf(varDataVal(tagTypeStr) + tagTypeLen, tagStrBufflen, "(%d)",
(int32_t)((*smrSuperTable).me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE)); (int32_t)((*smrSuperTable).me.stbEntry.schemaTag.pSchema[i].bytes - VARSTR_HEADER_SIZE));
} }
varDataSetLen(tagTypeStr, tagTypeLen); varDataSetLen(tagTypeStr, tagTypeLen);
@ -1127,7 +1134,7 @@ static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo,
QUERY_CHECK_NULL(tagVarChar, code, lino, _end, terrno); QUERY_CHECK_NULL(tagVarChar, code, lino, _end, terrno);
int32_t len = -1; int32_t len = -1;
if (tagLen > 0) if (tagLen > 0)
convertTagDataToStr(varDataVal(tagVarChar), tagType, tagData, tagLen, &len); convertTagDataToStr(varDataVal(tagVarChar), bufSize + 1 - VARSTR_HEADER_SIZE, tagType, tagData, tagLen, &len);
else else
len = 0; len = 0;
varDataSetLen(tagVarChar, len); varDataSetLen(tagVarChar, len);
@ -1197,13 +1204,19 @@ static int32_t sysTableUserColsFillOneTableCols(const SSysTableScanInfo* pInfo,
int8_t colType = schemaRow->pSchema[i].type; int8_t colType = schemaRow->pSchema[i].type;
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 4); pColInfoData = taosArrayGet(dataBlock->pDataBlock, 4);
QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno); QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno);
int32_t colStrBufflen = 32;
char colTypeStr[VARSTR_HEADER_SIZE + 32]; char colTypeStr[VARSTR_HEADER_SIZE + 32];
int colTypeLen = sprintf(varDataVal(colTypeStr), "%s", tDataTypes[colType].name); int colTypeLen = tsnprintf(varDataVal(colTypeStr), colStrBufflen, "%s", tDataTypes[colType].name);
colStrBufflen -= colTypeLen;
if (colStrBufflen <= 0) {
code = TSDB_CODE_INVALID_PARA;
QUERY_CHECK_CODE(code, lino, _end);
}
if (colType == TSDB_DATA_TYPE_VARCHAR) { if (colType == TSDB_DATA_TYPE_VARCHAR) {
colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d)", colTypeLen += tsnprintf(varDataVal(colTypeStr) + colTypeLen, colStrBufflen, "(%d)",
(int32_t)(schemaRow->pSchema[i].bytes - VARSTR_HEADER_SIZE)); (int32_t)(schemaRow->pSchema[i].bytes - VARSTR_HEADER_SIZE));
} else if (colType == TSDB_DATA_TYPE_NCHAR) { } else if (colType == TSDB_DATA_TYPE_NCHAR) {
colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d)", colTypeLen += tsnprintf(varDataVal(colTypeStr) + colTypeLen, colStrBufflen, "(%d)",
(int32_t)((schemaRow->pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)); (int32_t)((schemaRow->pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
} }
varDataSetLen(colTypeStr, colTypeLen); varDataSetLen(colTypeStr, colTypeLen);
@ -2019,8 +2032,7 @@ static EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) {
SValueNode* node = (SValueNode*)pNode; SValueNode* node = (SValueNode*)pNode;
char* dbName = nodesGetValueFromNode(node); char* dbName = nodesGetValueFromNode(node);
strncpy(pContext, varDataVal(dbName), varDataLen(dbName)); tstrncpy((char*)pContext, varDataVal(dbName), TSDB_DB_NAME_LEN);
*((char*)pContext + varDataLen(dbName)) = 0;
return DEAL_RES_END; // stop walk return DEAL_RES_END; // stop walk
} }
default: default:
@ -2056,11 +2068,11 @@ static int32_t doSysTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes)
getDBNameFromCondition(pInfo->pCondition, dbName); getDBNameFromCondition(pInfo->pCondition, dbName);
if (strncasecmp(name, TSDB_INS_TABLE_COMPACTS, TSDB_TABLE_FNAME_LEN) != 0 && if (strncasecmp(name, TSDB_INS_TABLE_COMPACTS, TSDB_TABLE_FNAME_LEN) != 0 &&
strncasecmp(name, TSDB_INS_TABLE_COMPACT_DETAILS, TSDB_TABLE_FNAME_LEN) != 0) { strncasecmp(name, TSDB_INS_TABLE_COMPACT_DETAILS, TSDB_TABLE_FNAME_LEN) != 0) {
sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName); TAOS_UNUSED(tsnprintf(pInfo->req.db, sizeof(pInfo->req.db), "%d.%s", pInfo->accountId, dbName));
} }
} else if (strncasecmp(name, TSDB_INS_TABLE_COLS, TSDB_TABLE_FNAME_LEN) == 0) { } else if (strncasecmp(name, TSDB_INS_TABLE_COLS, TSDB_TABLE_FNAME_LEN) == 0) {
getDBNameFromCondition(pInfo->pCondition, dbName); getDBNameFromCondition(pInfo->pCondition, dbName);
if (dbName[0]) sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName); if (dbName[0]) TAOS_UNUSED(tsnprintf(pInfo->req.db, sizeof(pInfo->req.db), "%d.%s", pInfo->accountId, dbName));
(void)sysTableIsCondOnOneTable(pInfo->pCondition, pInfo->req.filterTb); (void)sysTableIsCondOnOneTable(pInfo->pCondition, pInfo->req.filterTb);
} }

4
source/libs/executor/test/executorTests.cpp
View File

@ -115,7 +115,7 @@ SSDataBlock* getDummyBlock(SOperatorInfo* pOperator) {
int32_t code = colDataSetVal(pColInfo, i, reinterpret_cast<const char*>(&v), false); int32_t code = colDataSetVal(pColInfo, i, reinterpret_cast<const char*>(&v), false);
ASSERT(code == 0); ASSERT(code == 0);
// sprintf(buf, "this is %d row", i); // tsnprintf(buf, "this is %d row", i);
// STR_TO_VARSTR(b1, buf); // STR_TO_VARSTR(b1, buf);
// //
// SColumnInfoData* pColInfo2 = static_cast<SColumnInfoData*>(TARRAY_GET_ELEM(pBlock->pDataBlock, 1)); // SColumnInfoData* pColInfo2 = static_cast<SColumnInfoData*>(TARRAY_GET_ELEM(pBlock->pDataBlock, 1));
@ -179,7 +179,7 @@ SSDataBlock* get2ColsDummyBlock(SOperatorInfo* pOperator) {
code = colDataSetVal(pColInfo1, i, reinterpret_cast<const char*>(&v), false); code = colDataSetVal(pColInfo1, i, reinterpret_cast<const char*>(&v), false);
ASSERT(code == 0); ASSERT(code == 0);
// sprintf(buf, "this is %d row", i); // tsnprintf(buf, "this is %d row", i);
// STR_TO_VARSTR(b1, buf); // STR_TO_VARSTR(b1, buf);
// //
// SColumnInfoData* pColInfo2 = static_cast<SColumnInfoData*>(TARRAY_GET_ELEM(pBlock->pDataBlock, 1)); // SColumnInfoData* pColInfo2 = static_cast<SColumnInfoData*>(TARRAY_GET_ELEM(pBlock->pDataBlock, 1));

2
source/libs/executor/test/lhashTests.cpp
View File

@ -26,7 +26,7 @@
TEST(testCase, linear_hash_Tests) { TEST(testCase, linear_hash_Tests) {
taosSeedRand(taosGetTimestampSec()); taosSeedRand(taosGetTimestampSec());
strcpy(tsTempDir, "/tmp/"); tstrncpy((char*)tsTempDir, "/tmp/", sizeof(tsTempDir));
_hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT); _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);

6
source/libs/monitor/src/monFramework.c
View File

@ -183,7 +183,7 @@ void monGenClusterInfoTable(SMonInfo *pMonitor){
} }
if (taosHashRemove(tsMonitor.metrics, metric_names[i], strlen(metric_names[i])) != 0) { if (taosHashRemove(tsMonitor.metrics, metric_names[i], strlen(metric_names[i])) != 0) {
uError("failed to remove metric %s", metric_names[i]); uTrace("failed to remove metric %s", metric_names[i]);
} }
} }
@ -652,7 +652,7 @@ void monGenMnodeRoleTable(SMonInfo *pMonitor){
} }
if (taosHashRemove(tsMonitor.metrics, mnodes_role_gauges[i], strlen(mnodes_role_gauges[i])) != 0) { if (taosHashRemove(tsMonitor.metrics, mnodes_role_gauges[i], strlen(mnodes_role_gauges[i])) != 0) {
uError("failed to remove metric %s", mnodes_role_gauges[i]); uTrace("failed to remove metric %s", mnodes_role_gauges[i]);
} }
} }
@ -725,7 +725,7 @@ void monGenVnodeRoleTable(SMonInfo *pMonitor){
} }
if (taosHashRemove(tsMonitor.metrics, vnodes_role_gauges[i], strlen(vnodes_role_gauges[i])) != 0) { if (taosHashRemove(tsMonitor.metrics, vnodes_role_gauges[i], strlen(vnodes_role_gauges[i])) != 0) {
uError("failed to remove metric %s", vnodes_role_gauges[i]); uTrace("failed to remove metric %s", vnodes_role_gauges[i]);
} }
} }

5
source/libs/monitor/test/monTest.cpp
View File

@ -26,7 +26,10 @@ class MonitorTest : public ::testing::Test {
monInit(&cfg); monInit(&cfg);
} }
static void TearDownTestSuite() { monCleanup(); } static void TearDownTestSuite() {
monCleanup();
taosMsleep(100);
}
public: public:
void SetUp() override {} void SetUp() override {}

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

@ -1974,14 +1974,13 @@ static SNode* setDatabaseOptionImpl(SAstCreateContext* pCxt, SNode* pOptions, ED
case DB_OPTION_S3_COMPACT: case DB_OPTION_S3_COMPACT:
pDbOptions->s3Compact = taosStr2Int8(((SToken*)pVal)->z, NULL, 10); pDbOptions->s3Compact = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
break; break;
case DB_OPTION_KEEP_TIME_OFFSET: { case DB_OPTION_KEEP_TIME_OFFSET:
pDbOptions->keepTimeOffset = taosStr2Int32(((SToken*)pVal)->z, NULL, 10); pDbOptions->keepTimeOffset = taosStr2Int32(((SToken*)pVal)->z, NULL, 10);
break; break;
case DB_OPTION_ENCRYPT_ALGORITHM: case DB_OPTION_ENCRYPT_ALGORITHM:
COPY_STRING_FORM_STR_TOKEN(pDbOptions->encryptAlgorithmStr, (SToken*)pVal); COPY_STRING_FORM_STR_TOKEN(pDbOptions->encryptAlgorithmStr, (SToken*)pVal);
pDbOptions->encryptAlgorithm = TSDB_DEFAULT_ENCRYPT_ALGO; pDbOptions->encryptAlgorithm = TSDB_DEFAULT_ENCRYPT_ALGO;
break; break;
}
default: default:
break; break;
} }

2
source/libs/qworker/src/qwUtil.c
View File

@ -533,7 +533,7 @@ int32_t qwSaveTbVersionInfo(qTaskInfo_t pTaskInfo, SQWTaskCtx *ctx) {
while (true) { while (true) {
tbGet = false; tbGet = false;
code = qGetQueryTableSchemaVersion(pTaskInfo, dbFName, tbName, &tbInfo.sversion, &tbInfo.tversion, i, &tbGet); code = qGetQueryTableSchemaVersion(pTaskInfo, dbFName, TSDB_DB_FNAME_LEN, tbName, TSDB_TABLE_NAME_LEN, &tbInfo.sversion, &tbInfo.tversion, i, &tbGet);
if (TSDB_CODE_SUCCESS != code || !tbGet) { if (TSDB_CODE_SUCCESS != code || !tbGet) {
break; break;
} }

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

@ -120,7 +120,7 @@ SStreamState* streamStateOpen(const char* path, void* pTask, int64_t streamId, i
SStreamTask* pStreamTask = pTask; SStreamTask* pStreamTask = pTask;
pState->streamId = streamId; pState->streamId = streamId;
pState->taskId = taskId; pState->taskId = taskId;
sprintf(pState->pTdbState->idstr, "0x%" PRIx64 "-0x%x", pState->streamId, pState->taskId); TAOS_UNUSED(tsnprintf(pState->pTdbState->idstr, sizeof(pState->pTdbState->idstr), "0x%" PRIx64 "-0x%x", pState->streamId, pState->taskId));
code = streamTaskSetDb(pStreamTask->pMeta, pTask, pState->pTdbState->idstr); code = streamTaskSetDb(pStreamTask->pMeta, pTask, pState->pTdbState->idstr);
QUERY_CHECK_CODE(code, lino, _end); QUERY_CHECK_CODE(code, lino, _end);

8
source/libs/stream/src/tstreamFileState.c
View File

@ -864,7 +864,7 @@ _end:
int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) { int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) {
char keyBuf[128] = {0}; char keyBuf[128] = {0};
sprintf(keyBuf, "%s:%" PRId64 "", TASK_KEY, checkpointId); TAOS_UNUSED(tsnprintf(keyBuf, sizeof(keyBuf), "%s:%" PRId64 "", TASK_KEY, checkpointId));
return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf); return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf);
} }
@ -886,14 +886,14 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
} }
memcpy(buf, val, len); memcpy(buf, val, len);
buf[len] = 0; buf[len] = 0;
maxCheckPointId = atol((char*)buf); maxCheckPointId = taosStr2Int64((char*)buf, NULL, 10);
taosMemoryFree(val); taosMemoryFree(val);
} }
for (int64_t i = maxCheckPointId; i > 0; i--) { for (int64_t i = maxCheckPointId; i > 0; i--) {
char buf[128] = {0}; char buf[128] = {0};
void* val = 0; void* val = 0;
int32_t len = 0; int32_t len = 0;
sprintf(buf, "%s:%" PRId64 "", TASK_KEY, i); TAOS_UNUSED(tsnprintf(buf, sizeof(buf), "%s:%" PRId64 "", TASK_KEY, i));
code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len); code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
if (code != 0) { if (code != 0) {
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
@ -903,7 +903,7 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
taosMemoryFree(val); taosMemoryFree(val);
TSKEY ts; TSKEY ts;
ts = atol((char*)buf); ts = taosStr2Int64((char*)buf, NULL, 10);
if (ts < mark) { if (ts < mark) {
// statekey winkey.ts < mark // statekey winkey.ts < mark
int32_t tmpRes = forceRemoveCheckpoint(pFileState, i); int32_t tmpRes = forceRemoveCheckpoint(pFileState, i);

15
source/libs/sync/src/syncAppendEntries.c
View File

@ -95,6 +95,8 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
bool accepted = false; bool accepted = false;
SSyncRaftEntry* pEntry = NULL; SSyncRaftEntry* pEntry = NULL;
bool resetElect = false; bool resetElect = false;
const STraceId* trace = &pRpcMsg->info.traceId;
char tbuf[40] = {0};
// if already drop replica, do not process // if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) { if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) {
@ -150,10 +152,10 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
goto _IGNORE; goto _IGNORE;
} }
sTrace("vgId:%d, recv append entries msg. index:%" PRId64 ", term:%" PRId64 ", preLogIndex:%" PRId64 sGTrace("vgId:%d, recv append entries msg. index:%" PRId64 ", term:%" PRId64 ", preLogIndex:%" PRId64
", prevLogTerm:%" PRId64 " commitIndex:%" PRId64 " entryterm:%" PRId64, ", prevLogTerm:%" PRId64 " commitIndex:%" PRId64 " entryterm:%" PRId64,
pMsg->vgId, pMsg->prevLogIndex + 1, pMsg->term, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->commitIndex, pMsg->vgId, pMsg->prevLogIndex + 1, pMsg->term, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->commitIndex,
pEntry->term); pEntry->term);
if (ths->fsmState == SYNC_FSM_STATE_INCOMPLETE) { if (ths->fsmState == SYNC_FSM_STATE_INCOMPLETE) {
pReply->fsmState = ths->fsmState; pReply->fsmState = ths->fsmState;
@ -179,6 +181,11 @@ _SEND_RESPONSE:
sTrace("vgId:%d, update commit return index %" PRId64 "", ths->vgId, returnIndex); sTrace("vgId:%d, update commit return index %" PRId64 "", ths->vgId, returnIndex);
} }
TRACE_SET_MSGID(&(rpcRsp.info.traceId), tGenIdPI64());
trace = &(rpcRsp.info.traceId);
sGTrace("vgId:%d, send append reply matchIndex:%" PRId64 " term:%" PRId64 " lastSendIndex:%" PRId64
" to dest: 0x%016" PRIx64,
ths->vgId, pReply->matchIndex, pReply->term, pReply->lastSendIndex, pReply->destId.addr);
// ack, i.e. send response // ack, i.e. send response
TAOS_CHECK_RETURN(syncNodeSendMsgById(&pReply->destId, ths, &rpcRsp)); TAOS_CHECK_RETURN(syncNodeSendMsgById(&pReply->destId, ths, &rpcRsp));

6
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -43,6 +43,8 @@ int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
int32_t code = 0; int32_t code = 0;
SyncAppendEntriesReply* pMsg = (SyncAppendEntriesReply*)pRpcMsg->pCont; SyncAppendEntriesReply* pMsg = (SyncAppendEntriesReply*)pRpcMsg->pCont;
int32_t ret = 0; int32_t ret = 0;
const STraceId* trace = &pRpcMsg->info.traceId;
char tbuf[40] = {0};
// if already drop replica, do not process // if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) { if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) {
@ -63,8 +65,8 @@ int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
return TSDB_CODE_SYN_WRONG_TERM; return TSDB_CODE_SYN_WRONG_TERM;
} }
sTrace("vgId:%d, received append entries reply. srcId:0x%016" PRIx64 ", term:%" PRId64 ", matchIndex:%" PRId64 "", sGTrace("vgId:%d, received append entries reply. srcId:0x%016" PRIx64 ", term:%" PRId64 ", matchIndex:%" PRId64 "",
pMsg->vgId, pMsg->srcId.addr, pMsg->term, pMsg->matchIndex); pMsg->vgId, pMsg->srcId.addr, pMsg->term, pMsg->matchIndex);
if (pMsg->success) { if (pMsg->success) {
SyncIndex oldMatchIndex = syncIndexMgrGetIndex(ths->pMatchIndex, &(pMsg->srcId)); SyncIndex oldMatchIndex = syncIndexMgrGetIndex(ths->pMatchIndex, &(pMsg->srcId));

24
source/libs/sync/src/syncPipeline.c
View File

@ -1026,6 +1026,14 @@ int32_t syncLogReplRecover(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEn
int32_t code = 0; int32_t code = 0;
if (pMgr->restored != false) return TSDB_CODE_SYN_INTERNAL_ERROR; if (pMgr->restored != false) return TSDB_CODE_SYN_INTERNAL_ERROR;
sTrace("vgId:%d, begin to recover sync log repl. peer: dnode:%d (%" PRIx64 "), repl-mgr:[%" PRId64 ", %" PRId64
", %" PRId64 ") restore:%d, buffer: [%" PRId64 ", %" PRId64 ", %" PRId64 ", %" PRId64
"), msg: {lastSendIndex:%" PRId64 ", matchIndex:%" PRId64 ", fsmState:%d, success:%d, lastMatchTerm:%" PRId64
"}",
pNode->vgId, DID(&destId), destId.addr, pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex, pMgr->restored,
pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex, pMsg->lastSendIndex, pMsg->matchIndex,
pMsg->fsmState, pMsg->success, pMsg->lastMatchTerm);
if (pMgr->endIndex == 0) { if (pMgr->endIndex == 0) {
if (pMgr->startIndex != 0) return TSDB_CODE_SYN_INTERNAL_ERROR; if (pMgr->startIndex != 0) return TSDB_CODE_SYN_INTERNAL_ERROR;
if (pMgr->matchIndex != 0) return TSDB_CODE_SYN_INTERNAL_ERROR; if (pMgr->matchIndex != 0) return TSDB_CODE_SYN_INTERNAL_ERROR;
@ -1171,6 +1179,11 @@ int32_t syncLogReplProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex inde
int64_t nowMs = taosGetMonoTimestampMs(); int64_t nowMs = taosGetMonoTimestampMs();
int32_t code = 0; int32_t code = 0;
sTrace("vgId:%d, begin to probe peer:%" PRIx64 " with msg of index:%" PRId64 ". repl-mgr:[%" PRId64 ", %" PRId64
", %" PRId64 "), restored:%d",
pNode->vgId, pNode->replicasId[pMgr->peerId].addr, index, pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex,
pMgr->restored);
if (pMgr->endIndex > pMgr->startIndex && if (pMgr->endIndex > pMgr->startIndex &&
nowMs < pMgr->states[pMgr->startIndex % pMgr->size].timeMs + retryMaxWaitMs) { nowMs < pMgr->states[pMgr->startIndex % pMgr->size].timeMs + retryMaxWaitMs) {
return 0; return 0;
@ -1206,6 +1219,10 @@ int32_t syncLogReplProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex inde
int32_t syncLogReplAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode) { int32_t syncLogReplAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
if (!pMgr->restored) return TSDB_CODE_SYN_INTERNAL_ERROR; if (!pMgr->restored) return TSDB_CODE_SYN_INTERNAL_ERROR;
sTrace("vgId:%d, begin to attempt replicate log entries from end to match. repl-mgr:[%" PRId64 ", %" PRId64
", %" PRId64 "), restore:%d",
pNode->vgId, pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex, pMgr->restored);
SRaftId* pDestId = &pNode->replicasId[pMgr->peerId]; SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
int32_t batchSize = TMAX(1, pMgr->size >> (4 + pMgr->retryBackoff)); int32_t batchSize = TMAX(1, pMgr->size >> (4 + pMgr->retryBackoff));
int32_t code = 0; int32_t code = 0;
@ -1527,11 +1544,12 @@ int32_t syncLogReplSendTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex ind
goto _err; goto _err;
} }
TRACE_SET_MSGID(&(msgOut.info.traceId), tGenIdPI64());
STraceId* trace = &(msgOut.info.traceId);
sGTrace("vgId:%d, replicate one msg index:%" PRId64 " term:%" PRId64 " prevterm:%" PRId64 " to dest: 0x%016" PRIx64,
pNode->vgId, pEntry->index, pEntry->term, prevLogTerm, pDestId->addr);
TAOS_CHECK_GOTO(syncNodeSendAppendEntries(pNode, pDestId, &msgOut), &lino, _err); TAOS_CHECK_GOTO(syncNodeSendAppendEntries(pNode, pDestId, &msgOut), &lino, _err);
sTrace("vgId:%d, replicate one msg index:%" PRId64 " term:%" PRId64 " prevterm:%" PRId64 " to dest: 0x%016" PRIx64,
pNode->vgId, pEntry->index, pEntry->term, prevLogTerm, pDestId->addr);
if (!inBuf) { if (!inBuf) {
syncEntryDestroy(pEntry); syncEntryDestroy(pEntry);
pEntry = NULL; pEntry = NULL;

4
source/libs/sync/src/syncUtil.c
View File

@ -152,8 +152,8 @@ static void syncLogReplStates2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLe
for (int32_t i = 0; i < pSyncNode->replicaNum; i++) { for (int32_t i = 0; i < pSyncNode->replicaNum; i++) {
SSyncLogReplMgr* pMgr = pSyncNode->logReplMgrs[i]; SSyncLogReplMgr* pMgr = pSyncNode->logReplMgrs[i];
if (pMgr == NULL) break; if (pMgr == NULL) break;
len += tsnprintf(buf + len, bufLen - len, "%d:%d [%" PRId64 " %" PRId64 ", %" PRId64 "]", i, pMgr->restored, len += tsnprintf(buf + len, bufLen - len, "%d:%d [%" PRId64 ", %" PRId64 ", %" PRId64 "]", i, pMgr->restored,
pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex); pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex);
if (i + 1 < pSyncNode->replicaNum) { if (i + 1 < pSyncNode->replicaNum) {
len += tsnprintf(buf + len, bufLen - len, "%s", ", "); len += tsnprintf(buf + len, bufLen - len, "%s", ", ");
} }

26
source/libs/transport/inc/transComm.h
View File

@ -278,19 +278,19 @@ bool transAsyncPoolIsEmpty(SAsyncPool* pool);
} \ } \
} while (0) } while (0)
#define ASYNC_CHECK_HANDLE(idMgt, id, exh1) \ #define ASYNC_CHECK_HANDLE(idMgt, id, exh1) \
do { \ do { \
if (id > 0) { \ if (id > 0) { \
SExHandle* exh2 = transAcquireExHandle(idMgt, id); \ SExHandle* exh2 = transAcquireExHandle(idMgt, id); \
if (exh2 == NULL || exh1 != exh2 || (exh2 != NULL && exh2->refId != id)) { \ if (exh2 == NULL || exh1 != exh2 || (exh2 != NULL && exh2->refId != id)) { \
tError("handle not match, exh1:%p, exh2:%p, refId:%"PRId64"", exh1, exh2, id); \ tDebug("handle not match, exh1:%p, exh2:%p, refId:%" PRId64 "", exh1, exh2, id); \
code = TSDB_CODE_INVALID_MSG; \ code = TSDB_CODE_INVALID_MSG; \
goto _return1; \ goto _return1; \
} \ } \
} else { \ } else { \
tError("invalid handle to release"); \ tDebug("invalid handle to release"); \
goto _return2; \ goto _return2; \
} \ } \
} while (0) } while (0)
int32_t transInitBuffer(SConnBuffer* buf); int32_t transInitBuffer(SConnBuffer* buf);

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

@ -3090,7 +3090,7 @@ int32_t transReleaseCliHandle(void* handle) {
static int32_t transInitMsg(void* pInstRef, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx, SCliReq** pCliMsg) { static int32_t transInitMsg(void* pInstRef, const SEpSet* pEpSet, STransMsg* pReq, STransCtx* ctx, SCliReq** pCliMsg) {
int32_t code = 0; int32_t code = 0;
TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64()); if (pReq->info.traceId.msgId == 0) TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64());
SCliReq* pCliReq = NULL; SCliReq* pCliReq = NULL;
SReqCtx* pCtx = taosMemoryCalloc(1, sizeof(SReqCtx)); SReqCtx* pCtx = taosMemoryCalloc(1, sizeof(SReqCtx));

64
source/util/src/tlog.c
View File

@ -21,10 +21,12 @@
#include "tjson.h" #include "tjson.h"
#include "tutil.h" #include "tutil.h"
#define LOG_MAX_LINE_SIZE (10024) #define LOG_MAX_LINE_SIZE (10024)
#define LOG_MAX_LINE_BUFFER_SIZE (LOG_MAX_LINE_SIZE + 3) #define LOG_MAX_LINE_BUFFER_SIZE (LOG_MAX_LINE_SIZE + 3)
#define LOG_MAX_LINE_DUMP_SIZE (1024 * 1024) #define LOG_MAX_STACK_LINE_SIZE (512)
#define LOG_MAX_LINE_DUMP_BUFFER_SIZE (LOG_MAX_LINE_DUMP_SIZE + 128) #define LOG_MAX_STACK_LINE_BUFFER_SIZE (LOG_MAX_STACK_LINE_SIZE + 3)
#define LOG_MAX_LINE_DUMP_SIZE (1024 * 1024)
#define LOG_MAX_LINE_DUMP_BUFFER_SIZE (LOG_MAX_LINE_DUMP_SIZE + 128)
#define LOG_FILE_DAY_LEN 64 #define LOG_FILE_DAY_LEN 64
@ -669,16 +671,40 @@ static inline void taosPrintLogImp(ELogLevel level, int32_t dflag, const char *b
} }
} }
void taosPrintLog(const char *flags, int32_t level, int32_t dflag, const char *format, ...) { /*
if (!(dflag & DEBUG_FILE) && !(dflag & DEBUG_SCREEN)) return; use taosPrintLogImpl_useStackBuffer to avoid stack overflow
char buffer[LOG_MAX_LINE_BUFFER_SIZE]; */
static int8_t taosPrintLogImplUseStackBuffer(const char *flags, int32_t level, int32_t dflag, const char *format,
va_list args) {
char buffer[LOG_MAX_STACK_LINE_BUFFER_SIZE];
int32_t len = taosBuildLogHead(buffer, flags); int32_t len = taosBuildLogHead(buffer, flags);
va_list argpointer; int32_t writeLen = len + vsnprintf(buffer + len, LOG_MAX_STACK_LINE_BUFFER_SIZE - len - 1, format, args);
va_start(argpointer, format); if (writeLen > LOG_MAX_STACK_LINE_SIZE) {
int32_t writeLen = len + vsnprintf(buffer + len, LOG_MAX_LINE_BUFFER_SIZE - len, format, argpointer); return 1;
va_end(argpointer); }
buffer[writeLen++] = '\n';
buffer[writeLen] = 0;
taosPrintLogImp(level, dflag, buffer, writeLen);
if (tsLogFp && level <= DEBUG_INFO) {
buffer[writeLen - 1] = 0;
(*tsLogFp)(taosGetTimestampMs(), level, buffer + len);
}
return 0;
}
static int8_t taosPrintLogImplUseHeapBuffer(const char *flags, int32_t level, int32_t dflag, const char *format,
va_list args) {
char *buffer = taosMemoryCalloc(1, LOG_MAX_LINE_BUFFER_SIZE + 1);
if (buffer == NULL) {
return 1;
}
int32_t len = taosBuildLogHead(buffer, flags);
int32_t writeLen = len + vsnprintf(buffer + len, LOG_MAX_LINE_BUFFER_SIZE - len - 1, format, args);
if (writeLen > LOG_MAX_LINE_SIZE) writeLen = LOG_MAX_LINE_SIZE; if (writeLen > LOG_MAX_LINE_SIZE) writeLen = LOG_MAX_LINE_SIZE;
buffer[writeLen++] = '\n'; buffer[writeLen++] = '\n';
@ -690,6 +716,22 @@ void taosPrintLog(const char *flags, int32_t level, int32_t dflag, const char *f
buffer[writeLen - 1] = 0; buffer[writeLen - 1] = 0;
(*tsLogFp)(taosGetTimestampMs(), level, buffer + len); (*tsLogFp)(taosGetTimestampMs(), level, buffer + len);
} }
taosMemoryFree(buffer);
return 0;
}
void taosPrintLog(const char *flags, int32_t level, int32_t dflag, const char *format, ...) {
if (!(dflag & DEBUG_FILE) && !(dflag & DEBUG_SCREEN)) return;
va_list argpointer, argpointer_copy;
va_start(argpointer, format);
va_copy(argpointer_copy, argpointer);
if (taosPrintLogImplUseStackBuffer(flags, level, dflag, format, argpointer) == 0) {
} else {
TAOS_UNUSED(taosPrintLogImplUseHeapBuffer(flags, level, dflag, format, argpointer_copy));
}
va_end(argpointer_copy);
va_end(argpointer);
} }
void taosPrintLongString(const char *flags, int32_t level, int32_t dflag, const char *format, ...) { void taosPrintLongString(const char *flags, int32_t level, int32_t dflag, const char *format, ...) {

7
source/util/test/CMakeLists.txt
View File

@ -126,6 +126,13 @@ add_test(
COMMAND regexTest COMMAND regexTest
) )
add_executable(logTest "log.cpp")
target_link_libraries(logTest os util common gtest_main)
add_test(
NAME logTest
COMMAND logTest
)
add_executable(decompressTest "decompressTest.cpp") add_executable(decompressTest "decompressTest.cpp")
target_link_libraries(decompressTest os util common gtest_main) target_link_libraries(decompressTest os util common gtest_main)
add_test( add_test(

46
source/util/test/log.cpp Normal file
View File

@ -0,0 +1,46 @@
#include <gtest/gtest.h>
#include <stdlib.h>
#include <time.h>
#include <random>
#include <tlog.h>
#include <iostream>
using namespace std;
TEST(log, check_log_refactor) {
const char *logDir = "/tmp";
const char *defaultLogFileNamePrefix = "taoslog";
const int32_t maxLogFileNum = 10000;
tsAsyncLog = 0;
// idxDebugFlag = 143;
strcpy(tsLogDir, (char *)logDir);
taosInitLog(tsLogDir, 10, false);
tsAsyncLog = 0;
uDebugFlag = 143;
std::string str;
str.push_back('a');
for (int i = 0; i < 10000; i += 2) {
str.push_back('a');
uError("write to file %s", str.c_str());
}
str.clear();
for (int i = 0; i < 10000; i += 2) {
str.push_back('a');
uDebug("write to file %s", str.c_str());
}
for (int i = 0; i < 10000; i += 2) {
str.push_back('a');
uInfo("write to file %s", str.c_str());
}
str.clear();
for (int i = 0; i < 10000; i += 2) {
str.push_back('a');
uTrace("write to file %s", str.c_str());
}
taosCloseLog();
}

2
tests/system-test/7-tmq/tmqVnodeTransform-stb-removewal.py
View File

@ -17,6 +17,8 @@ sys.path.append("./7-tmq")
from tmqCommon import * from tmqCommon import *
class TDTestCase: class TDTestCase:
updatecfgDict = {'sDebugFlag':143}
def __init__(self): def __init__(self):
self.vgroups = 1 self.vgroups = 1
self.ctbNum = 10 self.ctbNum = 10