参数绑定写入
- WebSocket 连接
- 原生连接

参数绑定写入

通过参数绑定方式写入数据时，能避免SQL语法解析的资源消耗，从而显著提升写入性能。参数绑定能提高写入效率的原因主要有以下几点：

减少解析时间：通过参数绑定，SQL 语句的结构在第一次执行时就已经确定，后续的执行只需要替换参数值，这样可以避免每次执行时都进行语法解析，从而减少解析时间。
预编译：当使用参数绑定时，SQL 语句可以被预编译并缓存，后续使用不同的参数值执行时，可以直接使用预编译的版本，提高执行效率。
减少网络开销：参数绑定还可以减少发送到数据库的数据量，因为只需要发送参数值而不是完整的 SQL 语句，特别是在执行大量相似的插入或更新操作时，这种差异尤为明显。

Tips: 数据写入推荐使用参数绑定方式

下面我们继续以智能电表为例，展示各语言连接器使用参数绑定高效写入的功能：

准备一个参数化的 SQL 插入语句，用于向超级表 meters 中插入数据。这个语句允许动态地指定子表名、标签和列值。
循环生成多个子表及其对应的数据行。对于每个子表：
设置子表的名称和标签值（分组 ID 和位置）。
生成多行数据，每行包括一个时间戳、随机生成的电流、电压和相位值。
执行批量插入操作，将这些数据行插入到对应的子表中。
最后打印实际插入表中的行数。

WebSocket 连接

Java
Python
Go
Rust
Node.js
C#
C
REST API

1. public class WSParameterBindingBasicDemo {

3. // modify host to your own
4. private static final String host = "127.0.0.1";
5. private static final Random random = new Random(System.currentTimeMillis());
6. private static final int numOfSubTable = 10, numOfRow = 10;

8. public static void main(String[] args) throws SQLException {

10. String jdbcUrl = "jdbc:TAOS-WS://" + host + ":6041";
11. try (Connection conn = DriverManager.getConnection(jdbcUrl, "root", "taosdata")) {
12. init(conn);

14. String sql = "INSERT INTO ? USING power.meters TAGS(?,?) VALUES (?,?,?,?)";

16. try (TSWSPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSWSPreparedStatement.class)) {

18. for (int i = 1; i <= numOfSubTable; i++) {
19. // set table name
20. pstmt.setTableName("d_bind_" + i);

22. // set tags
23. pstmt.setTagInt(0, i);
24. pstmt.setTagString(1, "location_" + i);

26. // set columns
27. long current = System.currentTimeMillis();
28. for (int j = 0; j < numOfRow; j++) {
29. pstmt.setTimestamp(1, new Timestamp(current + j));
30. pstmt.setFloat(2, random.nextFloat() * 30);
31. pstmt.setInt(3, random.nextInt(300));
32. pstmt.setFloat(4, random.nextFloat());
33. pstmt.addBatch();
34. }
35. int[] exeResult = pstmt.executeBatch();
36. // you can check exeResult here
37. System.out.println("Successfully inserted " + exeResult.length + " rows to power.meters.");
38. }
39. }
40. } catch (Exception ex) {
41. // please refer to the JDBC specifications for detailed exceptions info
42. System.out.printf("Failed to insert to table meters using stmt, %sErrMessage: %s%n",
43. ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
44. ex.getMessage());
45. // Print stack trace for context in examples. Use logging in production.
46. ex.printStackTrace();
47. throw ex;
48. }
49. }

51. private static void init(Connection conn) throws SQLException {
52. try (Statement stmt = conn.createStatement()) {
53. stmt.execute("CREATE DATABASE IF NOT EXISTS power");
54. stmt.execute("USE power");
55. stmt.execute(
56. "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))");
57. }
58. }
59. }

查看源码

这是一个更详细的参数绑定示例

1. from datetime import datetime
2. import random
3. import taosws

5. numOfSubTable = 10

7. numOfRow = 10

9. conn = None
10. stmt = None
11. host="localhost"
12. port=6041
13. try:
14. conn = taosws.connect(user="root",
15. password="taosdata",
16. host=host,
17. port=port)

19. conn.execute("CREATE DATABASE IF NOT EXISTS power")
20. conn.execute("USE power")
21. conn.execute(
22. "CREATE TABLE IF NOT EXISTS `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))"
23. )

25. # ANCHOR: stmt
26. sql = "INSERT INTO ? USING meters (groupid, location) TAGS(?,?) VALUES (?,?,?,?)"
27. stmt = conn.statement()
28. stmt.prepare(sql)

30. for i in range(numOfSubTable):
31. tbname = f"d_bind_{i}"

33. tags = [
34. taosws.int_to_tag(i),
35. taosws.varchar_to_tag(f"location_{i}"),
36. ]
37. stmt.set_tbname_tags(tbname, tags)
38. current = int(datetime.now().timestamp() * 1000)
39. timestamps = []
40. currents = []
41. voltages = []
42. phases = []
43. for j in range (numOfRow):
44. timestamps.append(current + i)
45. currents.append(random.random() * 30)
46. voltages.append(random.randint(100, 300))
47. phases.append(random.random())

49. stmt.bind_param(
50. [
51. taosws.millis_timestamps_to_column(timestamps),
52. taosws.floats_to_column(currents),
53. taosws.ints_to_column(voltages),
54. taosws.floats_to_column(phases),
55. ]
56. )

58. stmt.add_batch()
59. stmt.execute()

61. print(f"Successfully inserted to power.meters.")

63. except Exception as err:
64. print(f"Failed to insert to table meters using stmt, ErrMessage:{err}")
65. raise err
66. finally:
67. if stmt:
68. stmt.close()
69. if conn:
70. conn.close()

查看源码

1. package main

3. import (
4. "database/sql"
5. "fmt"
6. "log"
7. "math/rand"
8. "time"

10. "github.com/taosdata/driver-go/v3/common"
11. "github.com/taosdata/driver-go/v3/common/param"
12. _ "github.com/taosdata/driver-go/v3/taosRestful"
13. "github.com/taosdata/driver-go/v3/ws/stmt"
14. )

16. func main() {
17. host := "127.0.0.1"
18. numOfSubTable := 10
19. numOfRow := 10

21. taosDSN := fmt.Sprintf("root:taosdata@http(%s:6041)/", host)
22. db, err := sql.Open("taosRestful", taosDSN)
23. if err != nil {
24. log.Fatalln("Failed to connect to " + taosDSN + "; ErrMessage: " + err.Error())
25. }
26. defer db.Close()
27. // prepare database and table
28. _, err = db.Exec("CREATE DATABASE IF NOT EXISTS power")
29. if err != nil {
30. log.Fatalln("Failed to create database power, ErrMessage: " + err.Error())
31. }
32. _, err = db.Exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
33. if err != nil {
34. log.Fatalln("Failed to create stable power.meters, ErrMessage: " + err.Error())
35. }

37. config := stmt.NewConfig(fmt.Sprintf("ws://%s:6041", host), 0)
38. config.SetConnectUser("root")
39. config.SetConnectPass("taosdata")
40. config.SetConnectDB("power")
41. config.SetMessageTimeout(common.DefaultMessageTimeout)
42. config.SetWriteWait(common.DefaultWriteWait)

44. connector, err := stmt.NewConnector(config)
45. if err != nil {
46. log.Fatalln("Failed to create stmt connector,url: " + taosDSN + "; ErrMessage: " + err.Error())
47. }
48. // prepare statement
49. sql := "INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)"
50. stmt, err := connector.Init()
51. if err != nil {
52. log.Fatalln("Failed to init stmt, sql: " + sql + ", ErrMessage: " + err.Error())
53. }
54. err = stmt.Prepare(sql)
55. if err != nil {
56. log.Fatal("Failed to prepare sql, sql: " + sql + ", ErrMessage: " + err.Error())
57. }
58. for i := 1; i <= numOfSubTable; i++ {
59. tableName := fmt.Sprintf("d_bind_%d", i)
60. tags := param.NewParam(2).AddInt(i).AddBinary([]byte(fmt.Sprintf("location_%d", i)))
61. tagsType := param.NewColumnType(2).AddInt().AddBinary(24)
62. columnType := param.NewColumnType(4).AddTimestamp().AddFloat().AddInt().AddFloat()
63. // set tableName
64. err = stmt.SetTableName(tableName)
65. if err != nil {
66. log.Fatal("Failed to set table name, tableName: " + tableName + "; ErrMessage: " + err.Error())
67. }
68. // set tags
69. err = stmt.SetTags(tags, tagsType)
70. if err != nil {
71. log.Fatal("Failed to set tags, ErrMessage: " + err.Error())
72. }
73. // bind column data
74. current := time.Now()
75. for j := 0; j < numOfRow; j++ {
76. columnData := make([]*param.Param, 4)
77. columnData[0] = param.NewParam(1).AddTimestamp(current.Add(time.Millisecond*time.Duration(j)), common.PrecisionMilliSecond)
78. columnData[1] = param.NewParam(1).AddFloat(rand.Float32() * 30)
79. columnData[2] = param.NewParam(1).AddInt(rand.Intn(300))
80. columnData[3] = param.NewParam(1).AddFloat(rand.Float32())
81. err = stmt.BindParam(columnData, columnType)
82. if err != nil {
83. log.Fatal("Failed to bind params, ErrMessage: " + err.Error())
84. }
85. }
86. // add batch
87. err = stmt.AddBatch()
88. if err != nil {
89. log.Fatal("Failed to add batch, ErrMessage: " + err.Error())
90. }
91. // execute batch
92. err = stmt.Exec()
93. if err != nil {
94. log.Fatal("Failed to exec, ErrMessage: " + err.Error())
95. }
96. // get affected rows
97. affected := stmt.GetAffectedRows()
98. // you can check exeResult here
99. fmt.Printf("Successfully inserted %d rows to %s.\n", affected, tableName)
100. }
101. err = stmt.Close()
102. if err != nil {
103. log.Fatal("Failed to close stmt, ErrMessage: " + err.Error())
104. }
105. }

查看源码

1. use taos::*;

3. #[tokio::main]
4. async fn main() -> anyhow::Result<()> {
5. let dsn = "ws://";
6. let taos = TaosBuilder::from_dsn(dsn)?.build().await?;

8. taos.exec("DROP DATABASE IF EXISTS power").await?;
9. taos.create_database("power").await?;
10. taos.use_database("power").await?;
11. taos.exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))").await?;

13. let mut stmt = Stmt::init(&taos).await?;
14. stmt.prepare("INSERT INTO ? USING meters TAGS(?, ?) VALUES(?, ?, ?, ?)").await?;

16. const NUM_TABLES: usize = 10;
17. const NUM_ROWS: usize = 10;
18. for i in 0..NUM_TABLES {
19. let table_name = format!("d_bind_{}", i);
20. let tags = vec![Value::Int(i as i32), Value::VarChar(format!("location_{}", i).into())];

22. // set table name and tags for the prepared statement.
23. match stmt.set_tbname_tags(&table_name, &tags).await{
24. Ok(_) => {},
25. Err(err) => {
26. eprintln!("Failed to set table name and tags, table_name:{}, tags:{:?}, ErrMessage: {}", table_name, tags, err);
27. return Err(err.into());
28. }
29. }
30. for j in 0..NUM_ROWS {
31. let values = vec![
32. ColumnView::from_millis_timestamp(vec![1648432611249 + j as i64]),
33. ColumnView::from_floats(vec![10.3 + j as f32]),
34. ColumnView::from_ints(vec![219 + j as i32]),
35. ColumnView::from_floats(vec![0.31 + j as f32]),
36. ];
37. // bind values to the prepared statement.
38. match stmt.bind(&values).await{
39. Ok(_) => {},
40. Err(err) => {
41. eprintln!("Failed to bind values, values:{:?}, ErrMessage: {}", values, err);
42. return Err(err.into());
43. }
44. }
45. }

47. match stmt.add_batch().await{
48. Ok(_) => {},
49. Err(err) => {
50. eprintln!("Failed to add batch, ErrMessage: {}", err);
51. return Err(err.into());
52. }
53. }
54. }

56. // execute.
57. match stmt.execute().await{
58. Ok(affected_rows) => println!("Successfully inserted {} rows to power.meters.", affected_rows),
59. Err(err) => {
60. eprintln!("Failed to insert to table meters using stmt, ErrMessage: {}", err);
61. return Err(err.into());
62. }
63. }

65. Ok(())
66. }

查看源码

1. const taos = require("@tdengine/websocket");

3. let db = 'power';
4. let stable = 'meters';
5. let numOfSubTable = 10;
6. let numOfRow = 10;
7. let dsn = 'ws://localhost:6041'
8. function getRandomInt(min, max) {
9. min = Math.ceil(min);
10. max = Math.floor(max);
11. return Math.floor(Math.random() * (max - min + 1)) + min;
12. }

14. async function prepare() {

16. let conf = new taos.WSConfig(dsn);
17. conf.setUser('root')
18. conf.setPwd('taosdata')
19. conf.setDb(db)
20. let wsSql = await taos.sqlConnect(conf);
21. await wsSql.exec(`CREATE DATABASE IF NOT EXISTS ${db} KEEP 3650 DURATION 10 BUFFER 16 WAL_LEVEL 1;`);
22. await wsSql.exec(`CREATE STABLE IF NOT EXISTS ${db}.${stable} (ts timestamp, current float, voltage int, phase float) TAGS (location binary(64), groupId int);`);
23. return wsSql
24. }

26. async function test() {
27. let stmt = null;
28. let connector = null;
29. try {
30. connector = await prepare();
31. stmt = await connector.stmtInit();
32. await stmt.prepare(`INSERT INTO ? USING ${db}.${stable} (location, groupId) TAGS (?, ?) VALUES (?, ?, ?, ?)`);
33. for (let i = 0; i < numOfSubTable; i++) {
34. await stmt.setTableName(`d_bind_${i}`);
35. let tagParams = stmt.newStmtParam();
36. tagParams.setVarchar([`location_${i}`]);
37. tagParams.setInt([i]);
38. await stmt.setTags(tagParams);
39. let timestampParams = [];
40. let currentParams = [];
41. let voltageParams = [];
42. let phaseParams = [];
43. const currentMillis = new Date().getTime();
44. for (let j = 0; j < numOfRow; j++) {
45. timestampParams.push(currentMillis + j);
46. currentParams.push(Math.random() * 30);
47. voltageParams.push(getRandomInt(100, 300));
48. phaseParams.push(Math.random());
49. }
50. let bindParams = stmt.newStmtParam();
51. bindParams.setTimestamp(timestampParams);
52. bindParams.setFloat(currentParams);
53. bindParams.setInt(voltageParams);
54. bindParams.setFloat(phaseParams);
55. await stmt.bind(bindParams);
56. await stmt.batch();
57. await stmt.exec();
58. console.log("Successfully inserted " + stmt.getLastAffected() + " to power.meters.");
59. }
60. }
61. catch (err) {
62. console.error(`Failed to insert to table meters using stmt, ErrCode: ${err.code}, ErrMessage: ${err.message}`);
63. throw err;
64. }
65. finally {
66. if (stmt) {
67. await stmt.close();
68. }
69. if (connector) {
70. await connector.close();
71. }
72. taos.destroy();
73. }
74. }

76. test()

查看源码

1. public static void Main(string[] args)
2. {
3. var host = "127.0.0.1";
4. var numOfSubTable = 10;
5. var numOfRow = 10;
6. var random = new Random();
7. var connectionString = $"protocol=WebSocket;host={host};port=6041;useSSL=false;username=root;password=taosdata";
8. try
9. {
10. var builder = new ConnectionStringBuilder(connectionString);
11. using (var client = DbDriver.Open(builder))
12. {
13. // create database
14. client.Exec("CREATE DATABASE IF NOT EXISTS power");
15. // use database
16. client.Exec("USE power");
17. // create table
18. client.Exec(
19. "CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))");
20. using (var stmt = client.StmtInit())
21. {
22. String sql = "INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)";
23. stmt.Prepare(sql);
24. for (int i = 1; i <= numOfSubTable; i++)
25. {
26. var tableName = $"d_bind_{i}";
27. // set table name
28. stmt.SetTableName(tableName);
29. // set tags
30. stmt.SetTags(new object[] { i, $"location_{i}" });
31. var current = DateTime.Now;
32. // bind rows
33. for (int j = 0; j < numOfRow; j++)
34. {
35. stmt.BindRow(new object[]
36. {
37. current.Add(TimeSpan.FromMilliseconds(j)),
38. random.NextSingle() * 30,
39. random.Next(300),
40. random.NextSingle()
41. });
42. }
43. // add batch
44. stmt.AddBatch();
45. // execute
46. stmt.Exec();
47. // get affected rows
48. var affectedRows = stmt.Affected();
49. Console.WriteLine($"Successfully inserted {affectedRows} rows to {tableName}.");
50. }
51. }
52. }
53. }
54. catch (TDengineError e)
55. {
56. // handle TDengine error
57. Console.WriteLine("Failed to insert to table meters using stmt, ErrCode: " + e.Code + ", ErrMessage: " + e.Error);
58. throw;
59. }
60. catch (Exception e)
61. {
62. // handle other exceptions
63. Console.WriteLine("Failed to insert to table meters using stmt, ErrMessage: " + e.Message);
64. throw;
65. }
66. }

查看源码

1. /*
2. * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3. *
4. * This program is free software: you can use, redistribute, and/or modify
5. * it under the terms of the GNU Affero General Public License, version 3
6. * or later ("AGPL"), as published by the Free Software Foundation.
7. *
8. * This program is distributed in the hope that it will be useful, but WITHOUT
9. * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10. * FITNESS FOR A PARTICULAR PURPOSE.
11. *
12. * You should have received a copy of the GNU Affero General Public License
13. * along with this program. If not, see <http://www.gnu.org/licenses/>.
14. */

16. // TAOS standard API example. The same syntax as MySQL, but only a subset
17. // to compile: gcc -o stmt_insert_demo stmt_insert_demo.c -ltaos

19. #include <stdio.h>
20. #include <stdlib.h>
21. #include <string.h>
22. #include <sys/time.h>
23. #include "taosws.h"

25. /**
26. * @brief execute sql only.
27. *
28. * @param taos
29. * @param sql
30. */
31. void executeSQL(WS_TAOS *taos, const char *sql) {
32. WS_RES *res = ws_query(taos, sql);
33. int     code = ws_errno(res);
34. if (code != 0) {
35. fprintf(stderr, "%s\n", ws_errstr(res));
36. ws_free_result(res);
37. ws_close(taos);
38. exit(EXIT_FAILURE);
39. }
40. ws_free_result(res);
41. }

43. /**
44. * @brief check return status and exit program when error occur.
45. *
46. * @param stmt
47. * @param code
48. * @param msg
49. */
50. void checkErrorCode(WS_STMT *stmt, int code, const char *msg) {
51. if (code != 0) {
52. fprintf(stderr, "%s. code: %d, error: %s\n", msg, code, ws_stmt_errstr(stmt));
53. ws_stmt_close(stmt);
54. exit(EXIT_FAILURE);
55. }
56. }

58. typedef struct {
59. int64_t ts;
60. float   current;
61. int     voltage;
62. float   phase;
63. } Row;

65. int num_of_sub_table = 10;
66. int num_of_row = 10;
67. int total_affected = 0;
68. /**
69. * @brief insert data using stmt API
70. *
71. * @param taos
72. */
73. void insertData(WS_TAOS *taos) {
74. // init
75. WS_STMT *stmt = ws_stmt_init(taos);
76. if (stmt == NULL) {
77. fprintf(stderr, "Failed to init ws_stmt, error: %s\n", ws_stmt_errstr(NULL));
78. exit(EXIT_FAILURE);
79. }
80. // prepare
81. const char *sql = "INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)";
82. int         code = ws_stmt_prepare(stmt, sql, 0);
83. checkErrorCode(stmt, code, "Failed to execute ws_stmt_prepare");
84. for (int i = 1; i <= num_of_sub_table; i++) {
85. char table_name[20];
86. sprintf(table_name, "d_bind_%d", i);
87. char location[20];
88. sprintf(location, "location_%d", i);

90. // set table name and tags
91. WS_MULTI_BIND tags[2];
92. // groupId
93. tags[0].buffer_type = TSDB_DATA_TYPE_INT;
94. tags[0].buffer_length = sizeof(int);
95. tags[0].length = (int32_t *)&tags[0].buffer_length;
96. tags[0].buffer = &i;
97. tags[0].is_null = NULL;
98. tags[0].num = 1;
99. // location
100. tags[1].buffer_type = TSDB_DATA_TYPE_BINARY;
101. tags[1].buffer_length = strlen(location);
102. tags[1].length = (int32_t *)&tags[1].buffer_length;
103. tags[1].buffer = location;
104. tags[1].is_null = NULL;
105. tags[1].num = 1;
106. code = ws_stmt_set_tbname_tags(stmt, table_name, tags, 2);
107. checkErrorCode(stmt, code, "Failed to set table name and tags\n");

109. // insert rows
110. WS_MULTI_BIND params[4];
111. // ts
112. params[0].buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
113. params[0].buffer_length = sizeof(int64_t);
114. params[0].length = (int32_t *)&params[0].buffer_length;
115. params[0].is_null = NULL;
116. params[0].num = 1;
117. // current
118. params[1].buffer_type = TSDB_DATA_TYPE_FLOAT;
119. params[1].buffer_length = sizeof(float);
120. params[1].length = (int32_t *)&params[1].buffer_length;
121. params[1].is_null = NULL;
122. params[1].num = 1;
123. // voltage
124. params[2].buffer_type = TSDB_DATA_TYPE_INT;
125. params[2].buffer_length = sizeof(int);
126. params[2].length = (int32_t *)&params[2].buffer_length;
127. params[2].is_null = NULL;
128. params[2].num = 1;
129. // phase
130. params[3].buffer_type = TSDB_DATA_TYPE_FLOAT;
131. params[3].buffer_length = sizeof(float);
132. params[3].length = (int32_t *)&params[3].buffer_length;
133. params[3].is_null = NULL;
134. params[3].num = 1;

136. for (int j = 0; j < num_of_row; j++) {
137. struct timeval tv;
138. gettimeofday(&tv, NULL);
139. long long milliseconds = tv.tv_sec * 1000LL + tv.tv_usec / 1000;  // current timestamp in milliseconds
140. int64_t   ts = milliseconds + j;
141. float     current = (float)rand() / RAND_MAX * 30;
142. int       voltage = rand() % 300;
143. float     phase = (float)rand() / RAND_MAX;
144. params[0].buffer = &ts;
145. params[1].buffer = &current;
146. params[2].buffer = &voltage;
147. params[3].buffer = &phase;
148. // bind param
149. code = ws_stmt_bind_param_batch(stmt, params, 4);
150. checkErrorCode(stmt, code, "Failed to bind param");
151. }
152. // add batch
153. code = ws_stmt_add_batch(stmt);
154. checkErrorCode(stmt, code, "Failed to add batch");
155. // execute batch
156. int affected_rows = 0;
157. code = ws_stmt_execute(stmt, &affected_rows);
158. checkErrorCode(stmt, code, "Failed to exec stmt");
159. // get affected rows
160. int affected = ws_stmt_affected_rows_once(stmt);
161. total_affected += affected;
162. }
163. fprintf(stdout, "Successfully inserted %d rows to power.meters.\n", total_affected);
164. ws_stmt_close(stmt);
165. }

167. int main() {
168. int      code = 0;
169. char    *dsn = "ws://localhost:6041";
170. WS_TAOS *taos = ws_connect(dsn);
171. if (taos == NULL) {
172. fprintf(stderr, "Failed to connect to %s, ErrCode: 0x%x, ErrMessage: %s.\n", dsn, ws_errno(NULL), ws_errstr(NULL));
173. exit(EXIT_FAILURE);
174. }
175. // create database and table
176. executeSQL(taos, "CREATE DATABASE IF NOT EXISTS power");
177. executeSQL(taos, "USE power");
178. executeSQL(taos,
179. "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS "
180. "(groupId INT, location BINARY(24))");
181. insertData(taos);
182. ws_close(taos);
183. }

查看源码

不支持

原生连接

Java
Python
Go
Rust
Node.js
C#
C
REST API

1. public class ParameterBindingBasicDemo {

3. // modify host to your own
4. private static final String host = "127.0.0.1";
5. private static final Random random = new Random(System.currentTimeMillis());
6. private static final int numOfSubTable = 10, numOfRow = 10;

8. public static void main(String[] args) throws SQLException {

10. String jdbcUrl = "jdbc:TAOS://" + host + ":6030/";
11. try (Connection conn = DriverManager.getConnection(jdbcUrl, "root", "taosdata")) {

13. init(conn);

15. String sql = "INSERT INTO ? USING power.meters TAGS(?,?) VALUES (?,?,?,?)";

17. try (TSDBPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class)) {

19. for (int i = 1; i <= numOfSubTable; i++) {
20. // set table name
21. pstmt.setTableName("d_bind_" + i);

23. // set tags
24. pstmt.setTagInt(0, i);
25. pstmt.setTagString(1, "location_" + i);

27. // set column ts
28. ArrayList<Long> tsList = new ArrayList<>();
29. long current = System.currentTimeMillis();
30. for (int j = 0; j < numOfRow; j++)
31. tsList.add(current + j);
32. pstmt.setTimestamp(0, tsList);

34. // set column current
35. ArrayList<Float> currentList = new ArrayList<>();
36. for (int j = 0; j < numOfRow; j++)
37. currentList.add(random.nextFloat() * 30);
38. pstmt.setFloat(1, currentList);

40. // set column voltage
41. ArrayList<Integer> voltageList = new ArrayList<>();
42. for (int j = 0; j < numOfRow; j++)
43. voltageList.add(random.nextInt(300));
44. pstmt.setInt(2, voltageList);

46. // set column phase
47. ArrayList<Float> phaseList = new ArrayList<>();
48. for (int j = 0; j < numOfRow; j++)
49. phaseList.add(random.nextFloat());
50. pstmt.setFloat(3, phaseList);
51. // add column
52. pstmt.columnDataAddBatch();
53. }
54. // execute column
55. pstmt.columnDataExecuteBatch();
56. // you can check exeResult here
57. System.out.println("Successfully inserted " + (numOfSubTable * numOfRow) + " rows to power.meters.");
58. }
59. } catch (Exception ex) {
60. // please refer to the JDBC specifications for detailed exceptions info
61. System.out.printf("Failed to insert to table meters using stmt, %sErrMessage: %s%n",
62. ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
63. ex.getMessage());
64. // Print stack trace for context in examples. Use logging in production.
65. ex.printStackTrace();
66. throw ex;
67. }
68. }

70. private static void init(Connection conn) throws SQLException {
71. try (Statement stmt = conn.createStatement()) {
72. stmt.execute("CREATE DATABASE IF NOT EXISTS power");
73. stmt.execute("USE power");
74. stmt.execute("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))");
75. }
76. }
77. }

查看源码

这是一个更详细的参数绑定示例

1. import taos
2. from datetime import datetime
3. import random

5. numOfSubTable = 10
6. numOfRow = 10

8. conn = None
9. stmt = None
10. host="localhost"
11. port=6030
12. try:
13. conn = taos.connect(
14. user="root",
15. password="taosdata",
16. host=host,
17. port=port,
18. )

20. conn.execute("CREATE DATABASE IF NOT EXISTS power")
21. conn.execute("USE power")
22. conn.execute(
23. "CREATE TABLE IF NOT EXISTS `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) TAGS (`groupid` INT, `location` BINARY(16))"
24. )

26. # ANCHOR: stmt
27. sql = "INSERT INTO ? USING meters (groupid, location) TAGS(?,?) VALUES (?,?,?,?)"
28. stmt = conn.statement(sql)
29. for i in range(numOfSubTable):
30. tbname = f"d_bind_{i}"

32. tags = taos.new_bind_params(2)
33. tags[0].int([i])
34. tags[1].binary([f"location_{i}"])
35. stmt.set_tbname_tags(tbname, tags)

37. current = int(datetime.now().timestamp() * 1000)
38. timestamps = []
39. currents = []
40. voltages = []
41. phases = []

43. for j in range (numOfRow):
44. timestamps.append(current + i)
45. currents.append(random.random() * 30)
46. voltages.append(random.randint(100, 300))
47. phases.append(random.random())

49. params = taos.new_bind_params(4)
50. params[0].timestamp(timestamps)
51. params[1].float(currents)
52. params[2].int(voltages)
53. params[3].float(phases)
54. stmt.bind_param_batch(params)
55. stmt.execute()
56. print(f"Successfully inserted to power.meters.")

58. except Exception as err:
59. print(f"Failed to insert to table meters using stmt, ErrMessage:{err}")
60. raise err
61. finally:
62. if stmt:
63. stmt.close()
64. if conn:
65. conn.close()

查看源码

1. package main

3. import (
4. "fmt"
5. "log"
6. "math/rand"
7. "time"

9. "github.com/taosdata/driver-go/v3/af"
10. "github.com/taosdata/driver-go/v3/common"
11. "github.com/taosdata/driver-go/v3/common/param"
12. )

14. func main() {
15. host := "127.0.0.1"
16. numOfSubTable := 10
17. numOfRow := 10
18. db, err := af.Open(host, "root", "taosdata", "", 0)
19. if err != nil {
20. log.Fatalln("Failed to connect to " + host + "; ErrMessage: " + err.Error())
21. }
22. defer db.Close()
23. // prepare database and table
24. _, err = db.Exec("CREATE DATABASE IF NOT EXISTS power")
25. if err != nil {
26. log.Fatalln("Failed to create database power, ErrMessage: " + err.Error())
27. }
28. _, err = db.Exec("USE power")
29. if err != nil {
30. log.Fatalln("Failed to use database power, ErrMessage: " + err.Error())
31. }
32. _, err = db.Exec("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))")
33. if err != nil {
34. log.Fatalln("Failed to create stable meters, ErrMessage: " + err.Error())
35. }
36. // prepare statement
37. sql := "INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)"
38. stmt := db.Stmt()
39. err = stmt.Prepare(sql)
40. if err != nil {
41. log.Fatalln("Failed to prepare sql, sql: " + sql + ", ErrMessage: " + err.Error())
42. }
43. for i := 1; i <= numOfSubTable; i++ {
44. tableName := fmt.Sprintf("d_bind_%d", i)
45. tags := param.NewParam(2).AddInt(i).AddBinary([]byte(fmt.Sprintf("location_%d", i)))
46. // set tableName and tags
47. err = stmt.SetTableNameWithTags(tableName, tags)
48. if err != nil {
49. log.Fatalln("Failed to set table name and tags, tableName: " + tableName + "; ErrMessage: " + err.Error())
50. }
51. // bind column data
52. current := time.Now()
53. for j := 0; j < numOfRow; j++ {
54. row := param.NewParam(4).
55. AddTimestamp(current.Add(time.Millisecond*time.Duration(j)), common.PrecisionMilliSecond).
56. AddFloat(rand.Float32() * 30).
57. AddInt(rand.Intn(300)).
58. AddFloat(rand.Float32())
59. err = stmt.BindRow(row)
60. if err != nil {
61. log.Fatalln("Failed to bind params, ErrMessage: " + err.Error())
62. }
63. }
64. // add batch
65. err = stmt.AddBatch()
66. if err != nil {
67. log.Fatalln("Failed to add batch, ErrMessage: " + err.Error())
68. }
69. // execute batch
70. err = stmt.Execute()
71. if err != nil {
72. log.Fatalln("Failed to exec, ErrMessage: " + err.Error())
73. }
74. // get affected rows
75. affected := stmt.GetAffectedRows()
76. // you can check exeResult here
77. fmt.Printf("Successfully inserted %d rows to %s.\n", affected, tableName)
78. }
79. err = stmt.Close()
80. if err != nil {
81. log.Fatal("failed to close statement, err:", err)
82. }
83. }

查看源码

1. use taos::*;

3. #[tokio::main]
4. async fn main() -> anyhow::Result<()> {
5. let dsn = "taos://localhost:6030";
6. let taos = TaosBuilder::from_dsn(dsn)?.build().await?;

8. taos.exec("DROP DATABASE IF EXISTS power").await?;
9. taos.create_database("power").await?;
10. taos.use_database("power").await?;
11. taos.exec("CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))").await?;

13. let mut stmt = Stmt::init(&taos).await?;
14. stmt.prepare("INSERT INTO ? USING meters TAGS(?, ?) VALUES(?, ?, ?, ?)").await?;

16. const NUM_TABLES: usize = 10;
17. const NUM_ROWS: usize = 10;
18. for i in 0..NUM_TABLES {
19. let table_name = format!("d_bind_{}", i);
20. let tags = vec![Value::Int(i as i32), Value::VarChar(format!("location_{}", i).into())];

22. // set table name and tags for the prepared statement.
23. match stmt.set_tbname_tags(&table_name, &tags).await{
24. Ok(_) => {},
25. Err(err) => {
26. eprintln!("Failed to set table name and tags, table_name:{}, tags:{:?}, ErrMessage: {}", table_name, tags, err);
27. return Err(err.into());
28. }
29. }
30. for j in 0..NUM_ROWS {
31. let values = vec![
32. ColumnView::from_millis_timestamp(vec![1648432611249 + j as i64]),
33. ColumnView::from_floats(vec![10.3 + j as f32]),
34. ColumnView::from_ints(vec![219 + j as i32]),
35. ColumnView::from_floats(vec![0.31 + j as f32]),
36. ];
37. // bind values to the prepared statement.
38. match stmt.bind(&values).await{
39. Ok(_) => {},
40. Err(err) => {
41. eprintln!("Failed to bind values, values:{:?}, ErrMessage: {}", values, err);
42. return Err(err.into());
43. }
44. }
45. }

47. match stmt.add_batch().await{
48. Ok(_) => {},
49. Err(err) => {
50. eprintln!("Failed to add batch, ErrMessage: {}", err);
51. return Err(err.into());
52. }
53. }
54. }

56. // execute.
57. match stmt.execute().await{
58. Ok(affected_rows) => println!("Successfully inserted {} rows to power.meters.", affected_rows),
59. Err(err) => {
60. eprintln!("Failed to insert to table meters using stmt, ErrMessage: {}", err);
61. return Err(err.into());
62. }
63. }

65. Ok(())
66. }

查看源码

不支持

1. public static void Main(string[] args)
2. {
3. var host = "127.0.0.1";
4. var numOfSubTable = 10;
5. var numOfRow = 10;
6. var random = new Random();
7. var connectionString = $"host={host};port=6030;username=root;password=taosdata";
8. try
9. {
10. var builder = new ConnectionStringBuilder(connectionString);
11. using (var client = DbDriver.Open(builder))
12. {
13. // create database
14. client.Exec("CREATE DATABASE IF NOT EXISTS power");
15. // use database
16. client.Exec("USE power");
17. // create table
18. client.Exec(
19. "CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))");
20. using (var stmt = client.StmtInit())
21. {
22. String sql = "INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)";
23. stmt.Prepare(sql);
24. for (int i = 1; i <= numOfSubTable; i++)
25. {
26. var tableName = $"d_bind_{i}";
27. // set table name
28. stmt.SetTableName(tableName);
29. // set tags
30. stmt.SetTags(new object[] { i, $"location_{i}" });
31. var current = DateTime.Now;
32. // bind rows
33. for (int j = 0; j < numOfRow; j++)
34. {
35. stmt.BindRow(new object[]
36. {
37. current.Add(TimeSpan.FromMilliseconds(j)),
38. random.NextSingle() * 30,
39. random.Next(300),
40. random.NextSingle()
41. });
42. }
43. // add batch
44. stmt.AddBatch();
45. // execute
46. stmt.Exec();
47. // get affected rows
48. var affectedRows = stmt.Affected();
49. Console.WriteLine($"Successfully inserted {affectedRows} rows to {tableName}.");
50. }
51. }
52. }
53. }
54. catch (TDengineError e)
55. {
56. // handle TDengine error
57. Console.WriteLine("Failed to insert to table meters using stmt, ErrCode: " + e.Code + ", ErrMessage: " + e.Error);
58. throw;
59. }
60. catch (Exception e)
61. {
62. // handle other exceptions
63. Console.WriteLine("Failed to insert to table meters using stmt, ErrMessage: " + e.Message);
64. throw;
65. }
66. }

查看源码

1. /*
2. * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3. *
4. * This program is free software: you can use, redistribute, and/or modify
5. * it under the terms of the GNU Affero General Public License, version 3
6. * or later ("AGPL"), as published by the Free Software Foundation.
7. *
8. * This program is distributed in the hope that it will be useful, but WITHOUT
9. * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10. * FITNESS FOR A PARTICULAR PURPOSE.
11. *
12. * You should have received a copy of the GNU Affero General Public License
13. * along with this program. If not, see <http://www.gnu.org/licenses/>.
14. */

16. // TAOS standard API example. The same syntax as MySQL, but only a subset
17. // to compile: gcc -o stmt_insert_demo stmt_insert_demo.c -ltaos

19. #include <stdio.h>
20. #include <stdlib.h>
21. #include <string.h>
22. #include <sys/time.h>
23. #include "taos.h"

25. /**
26. * @brief execute sql only.
27. *
28. * @param taos
29. * @param sql
30. */
31. void executeSQL(TAOS *taos, const char *sql) {
32. TAOS_RES *res = taos_query(taos, sql);
33. int       code = taos_errno(res);
34. if (code != 0) {
35. fprintf(stderr, "%s\n", taos_errstr(res));
36. taos_free_result(res);
37. taos_close(taos);
38. exit(EXIT_FAILURE);
39. }
40. taos_free_result(res);
41. }

43. /**
44. * @brief check return status and exit program when error occur.
45. *
46. * @param stmt
47. * @param code
48. * @param msg
49. */
50. void checkErrorCode(TAOS_STMT *stmt, int code, const char *msg) {
51. if (code != 0) {
52. fprintf(stderr, "%s. code: %d, error: %s\n", msg,code,taos_stmt_errstr(stmt));
53. taos_stmt_close(stmt);
54. exit(EXIT_FAILURE);
55. }
56. }

58. typedef struct {
59. int64_t ts;
60. float   current;
61. int     voltage;
62. float   phase;
63. } Row;

65. int num_of_sub_table = 10;
66. int num_of_row = 10;
67. int total_affected = 0;
68. /**
69. * @brief insert data using stmt API
70. *
71. * @param taos
72. */
73. void insertData(TAOS *taos) {
74. // init
75. TAOS_STMT *stmt = taos_stmt_init(taos);
76. if (stmt == NULL) {
77. fprintf(stderr, "Failed to init taos_stmt, error: %s\n", taos_stmt_errstr(NULL));
78. exit(EXIT_FAILURE);
79. }
80. // prepare
81. const char *sql = "INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)";
82. int         code = taos_stmt_prepare(stmt, sql, 0);
83. checkErrorCode(stmt, code, "Failed to execute taos_stmt_prepare");
84. for (int i = 1; i <= num_of_sub_table; i++) {
85. char table_name[20];
86. sprintf(table_name, "d_bind_%d", i);
87. char location[20];
88. sprintf(location, "location_%d", i);

90. // set table name and tags
91. TAOS_MULTI_BIND tags[2];
92. // groupId
93. tags[0].buffer_type = TSDB_DATA_TYPE_INT;
94. tags[0].buffer_length = sizeof(int);
95. tags[0].length = (int32_t *)&tags[0].buffer_length;
96. tags[0].buffer = &i;
97. tags[0].is_null = NULL;
98. tags[0].num = 1;
99. // location
100. tags[1].buffer_type = TSDB_DATA_TYPE_BINARY;
101. tags[1].buffer_length = strlen(location);
102. tags[1].length =(int32_t *) &tags[1].buffer_length;
103. tags[1].buffer = location;
104. tags[1].is_null = NULL;
105. tags[1].num = 1;
106. code = taos_stmt_set_tbname_tags(stmt, table_name, tags);
107. checkErrorCode(stmt, code, "Failed to set table name and tags\n");

109. // insert rows
110. TAOS_MULTI_BIND params[4];
111. // ts
112. params[0].buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
113. params[0].buffer_length = sizeof(int64_t);
114. params[0].length = (int32_t *)&params[0].buffer_length;
115. params[0].is_null = NULL;
116. params[0].num = 1;
117. // current
118. params[1].buffer_type = TSDB_DATA_TYPE_FLOAT;
119. params[1].buffer_length = sizeof(float);
120. params[1].length = (int32_t *)&params[1].buffer_length;
121. params[1].is_null = NULL;
122. params[1].num = 1;
123. // voltage
124. params[2].buffer_type = TSDB_DATA_TYPE_INT;
125. params[2].buffer_length = sizeof(int);
126. params[2].length = (int32_t *)&params[2].buffer_length;
127. params[2].is_null = NULL;
128. params[2].num = 1;
129. // phase
130. params[3].buffer_type = TSDB_DATA_TYPE_FLOAT;
131. params[3].buffer_length = sizeof(float);
132. params[3].length = (int32_t *)&params[3].buffer_length;
133. params[3].is_null = NULL;
134. params[3].num = 1;

136. for (int j = 0; j < num_of_row; j++) {
137. struct timeval tv;
138. gettimeofday(&tv, NULL);
139. long long milliseconds = tv.tv_sec * 1000LL + tv.tv_usec / 1000;  // current timestamp in milliseconds
140. int64_t   ts = milliseconds + j;
141. float     current = (float)rand() / RAND_MAX * 30;
142. int       voltage = rand() % 300;
143. float     phase = (float)rand() / RAND_MAX;
144. params[0].buffer = &ts;
145. params[1].buffer = &current;
146. params[2].buffer = &voltage;
147. params[3].buffer = &phase;
148. // bind param
149. code = taos_stmt_bind_param(stmt, params);
150. checkErrorCode(stmt, code, "Failed to bind param");
151. }
152. // add batch
153. code = taos_stmt_add_batch(stmt);
154. checkErrorCode(stmt, code, "Failed to add batch");
155. // execute batch
156. code = taos_stmt_execute(stmt);
157. checkErrorCode(stmt, code, "Failed to exec stmt");
158. // get affected rows
159. int affected = taos_stmt_affected_rows_once(stmt);
160. total_affected += affected;
161. }
162. fprintf(stdout, "Successfully inserted %d rows to power.meters.\n", total_affected);
163. taos_stmt_close(stmt);
164. }

166. int main() {
167. const char *host      = "localhost";
168. const char *user      = "root";
169. const char *password  = "taosdata";
170. uint16_t    port      = 6030;
171. TAOS *taos = taos_connect(host, user, password, NULL, port);
172. if (taos == NULL) {
173. fprintf(stderr, "Failed to connect to %s:%hu, ErrCode: 0x%x, ErrMessage: %s.\n", host, port, taos_errno(NULL), taos_errstr(NULL));
174. taos_cleanup();
175. exit(EXIT_FAILURE);
176. }
177. // create database and table
178. executeSQL(taos, "CREATE DATABASE IF NOT EXISTS power");
179. executeSQL(taos, "USE power");
180. executeSQL(taos,
181. "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS "
182. "(groupId INT, location BINARY(24))");
183. insertData(taos);
184. taos_close(taos);
185. taos_cleanup();
186. }

查看源码

不支持