我的书签
添加书签
移除书签Query Data
Introduction
SQL is used by TDengine as its query language. Application programs can send SQL statements to TDengine through REST API or connectors. TDengine’s CLI taos can also be used to execute ad hoc SQL queries. Here is the list of major query functionalities supported by TDengine：
- Query on single column or multiple columns
- Filter on tags or data columns：>, <, =, <>, like
- Grouping of results:
Group By- Sorting of results:Order By- Limit the number of results:Limit/Offset - Windowed aggregate queries for time windows (interval), session windows (session), and state windows (state_window)
- Arithmetic on columns of numeric types or aggregate results
- Join query with timestamp alignment
- Aggregate functions: count, max, min, avg, sum, twa, stddev, leastsquares, top, bottom, first, last, percentile, apercentile, last_row, spread, diff
For example, the SQL statement below can be executed in TDengine CLI taos to select records with voltage greater than 215 and limit the output to only 2 rows.
taos> select * from d1001 where voltage > 215 order by ts desc limit 2;ts | current | voltage | phase |======================================================================================2018-10-03 14:38:16.800 | 12.30000 | 221 | 0.31000 |2018-10-03 14:38:15.000 | 12.60000 | 218 | 0.33000 |Query OK, 2 row(s) in set (0.001100s)
To meet the requirements of varied use cases, some special functions have been added in TDengine. Some examples are twa (Time Weighted Average), spread (The difference between the maximum and the minimum), and last_row (the last row).
For detailed query syntax, see Select.
Aggregation among Tables
In most use cases, there are always multiple kinds of data collection points. A new concept, called STable (abbreviation for super table), is used in TDengine to represent one type of data collection point, and a subtable is used to represent a specific data collection point of that type. Tags are used by TDengine to represent the static properties of data collection points. A specific data collection point has its own values for static properties. By specifying filter conditions on tags, aggregation can be performed efficiently among all the subtables created via the same STable, i.e. same type of data collection points. Aggregate functions applicable for tables can be used directly on STables; the syntax is exactly the same.
Example 1
In TDengine CLI taos, use the SQL below to get the average voltage of all the meters in California grouped by location.
taos> SELECT AVG(voltage), location FROM meters GROUP BY location;avg(voltage) | location |===============================================================================================219.200000000 | California.SanFrancisco |221.666666667 | California.LosAngeles |Query OK, 2 rows in database (0.005995s)
Example 2
In TDengine CLI taos, use the SQL below to get the number of rows and the maximum current from meters whose groupId is 2.
taos> SELECT count(*), max(current) FROM meters where groupId = 2;count(*) | max(current) |==================================5 | 13.4 |Query OK, 1 row(s) in set (0.002136s)
In Select, all query operations are marked as to whether they support STables or not.
Down Sampling and Interpolation
In IoT use cases, down sampling is widely used to aggregate data by time range. The INTERVAL keyword in TDengine can be used to simplify the query by time window. For example, the SQL statement below can be used to get the sum of current every 10 seconds from meters table d1001.
taos> SELECT _wstart, sum(current) FROM d1001 INTERVAL(10s);_wstart | sum(current) |======================================================2018-10-03 14:38:00.000 | 10.300000191 |2018-10-03 14:38:10.000 | 24.900000572 |Query OK, 2 rows in database (0.003139s)
Down sampling can also be used for STable. For example, the below SQL statement can be used to get the sum of current from all meters in California.
taos> SELECT _wstart, SUM(current) FROM meters where location like "California%" INTERVAL(1s);_wstart | sum(current) |======================================================2018-10-03 14:38:04.000 | 10.199999809 |2018-10-03 14:38:05.000 | 23.699999809 |2018-10-03 14:38:06.000 | 11.500000000 |2018-10-03 14:38:15.000 | 12.600000381 |2018-10-03 14:38:16.000 | 34.400000572 |Query OK, 5 rows in database (0.007413s)
Down sampling also supports time offset. For example, the below SQL statement can be used to get the sum of current from all meters but each time window must start at the boundary of 500 milliseconds.
taos> SELECT _wstart, SUM(current) FROM meters INTERVAL(1s, 500a);_wstart | sum(current) |======================================================2018-10-03 14:38:03.500 | 10.199999809 |2018-10-03 14:38:04.500 | 10.300000191 |2018-10-03 14:38:05.500 | 13.399999619 |2018-10-03 14:38:06.500 | 11.500000000 |2018-10-03 14:38:14.500 | 12.600000381 |2018-10-03 14:38:16.500 | 34.400000572 |Query OK, 6 rows in database (0.005515s)
In many use cases, it’s hard to align the timestamp of the data collected by each collection point. However, a lot of algorithms like FFT require the data to be aligned with same time interval and application programs have to handle this by themselves. In TDengine, it’s easy to achieve the alignment using down sampling.
Interpolation can be performed in TDengine if there is no data in a time range.
For more information, see Aggregate by Window.
Examples
Query
In the section describing Insert, a database named power is created and some data are inserted into STable meters. Below sample code demonstrates how to query the data in this STable.
- Java
- Python
- Go
- Rust
- Node.js
- C#
- C
- PHP
package com.taos.example;import java.sql.*;public class RestQueryExample {private static Connection getConnection() throws SQLException {String jdbcUrl = "jdbc:TAOS-RS://localhost:6041/power?user=root&password=taosdata";return DriverManager.getConnection(jdbcUrl);}private static void printRow(ResultSet rs) throws SQLException {ResultSetMetaData meta = rs.getMetaData();for (int i = 1; i <= meta.getColumnCount(); i++) {String value = rs.getString(i);System.out.print(value);System.out.print("\t");}System.out.println();}private static void printColName(ResultSet rs) throws SQLException {ResultSetMetaData meta = rs.getMetaData();for (int i = 1; i <= meta.getColumnCount(); i++) {String colLabel = meta.getColumnLabel(i);System.out.print(colLabel);System.out.print("\t");}System.out.println();}private static void processResult(ResultSet rs) throws SQLException {printColName(rs);while (rs.next()) {printRow(rs);}}private static void queryData() throws SQLException {try (Connection conn = getConnection()) {try (Statement stmt = conn.createStatement()) {ResultSet rs = stmt.executeQuery("SELECT AVG(voltage) FROM meters GROUP BY location");processResult(rs);}}}public static void main(String[] args) throws SQLException {queryData();}}// possible output:// avg(voltage) location// 222.0 California.LosAngeles// 219.0 California.SanFrancisco
Result set is iterated row by row.
def query_api_demo(conn: taos.TaosConnection):result: taos.TaosResult = conn.query("SELECT tbname, * FROM meters LIMIT 2")print("field count:", result.field_count)print("meta of fields[1]:", result.fields[1])print("======================Iterate on result=========================")for row in result:print(row)# field count: 7# meta of fields[1]: {name: ts, type: 9, bytes: 8}# ======================Iterate on result=========================# ('d1003', datetime.datetime(2018, 10, 3, 14, 38, 5, 500000), 11.800000190734863, 221, 0.2800000011920929, 'california.losangeles', 2)# ('d1003', datetime.datetime(2018, 10, 3, 14, 38, 16, 600000), 13.399999618530273, 223, 0.28999999165534973, 'california.losangeles', 2)
Result set is retrieved as a whole, each row is converted to a dict and returned.
def fetch_all_demo(conn: taos.TaosConnection):result: taos.TaosResult = conn.query("SELECT ts, current FROM meters LIMIT 2")rows = result.fetch_all_into_dict()print("row count:", result.row_count)print("===============all data===================")print(rows)# row count: 2# ===============all data===================# [{'ts': datetime.datetime(2018, 10, 3, 14, 38, 5, 500000), 'current': 11.800000190734863},# {'ts': datetime.datetime(2018, 10, 3, 14, 38, 16, 600000), 'current': 13.399999618530273}]
package mainimport ("database/sql""log""time"_ "github.com/taosdata/driver-go/v3/taosRestful")func main() {var taosDSN = "root:taosdata@http(localhost:6041)/power"taos, err := sql.Open("taosRestful", taosDSN)if err != nil {log.Fatalln("failed to connect TDengine, err:", err)}defer taos.Close()rows, err := taos.Query("SELECT ts, current FROM meters LIMIT 2")if err != nil {log.Fatalln("failed to select from table, err:", err)}defer rows.Close()for rows.Next() {var r struct {ts time.Timecurrent float32}err := rows.Scan(&r.ts, &r.current)if err != nil {log.Fatalln("scan error:\n", err)return}log.Println(r.ts, r.current)}}
use taos::sync::*;fn main() -> anyhow::Result<()> {let taos = TaosBuilder::from_dsn("ws:///power")?.build()?;let mut result = taos.query("SELECT ts, current FROM meters LIMIT 2")?;// print column nameslet meta = result.fields();println!("{}", meta.iter().map(|field| field.name()).join("\t"));// print rowslet rows = result.rows();for row in rows {let row = row?;for (_name, value) in row {print!("{}\t", value);}println!();}Ok(())}// output(suppose you are in +8 timezone):// ts current// 2018-10-03T14:38:05+08:00 10.3// 2018-10-03T14:38:15+08:00 12.6
const taos = require("@tdengine/client");const conn = taos.connect({ host: "localhost", database: "power" });const cursor = conn.cursor();const query = cursor.query("SELECT ts, current FROM meters LIMIT 2");query.execute().then(function (result) {result.pretty();});// output:// Successfully connected to TDengine// ts | current |// =======================================================// 2018-10-03 14:38:05.000 | 10.3 |// 2018-10-03 14:38:15.000 | 12.6 |
using TDengineDriver;using TDengineDriver.Impl;using System.Runtime.InteropServices;namespace TDengineExample{internal class QueryExample{static void Main(){IntPtr conn = GetConnection();try{// run queryIntPtr res = TDengine.Query(conn, "SELECT * FROM meters LIMIT 2");if (TDengine.ErrorNo(res) != 0){throw new Exception("Failed to query since: " + TDengine.Error(res));}// get filed countint fieldCount = TDengine.FieldCount(res);Console.WriteLine("fieldCount=" + fieldCount);// print column namesList<TDengineMeta> metas = LibTaos.GetMeta(res);for (int i = 0; i < metas.Count; i++){Console.Write(metas[i].name + "\t");}Console.WriteLine();// print valuesList<Object> resData = LibTaos.GetData(res);for (int i = 0; i < resData.Count; i++){Console.Write($"|{resData[i].ToString()} \t");if (((i + 1) % metas.Count == 0)){Console.WriteLine("");}}Console.WriteLine();// Free result after useTDengine.FreeResult(res);}finally{TDengine.Close(conn);}}static IntPtr GetConnection(){string host = "localhost";short port = 6030;string username = "root";string password = "taosdata";string dbname = "power";var conn = TDengine.Connect(host, username, password, dbname, port);if (conn == IntPtr.Zero){throw new Exception("Connect to TDengine failed");}else{Console.WriteLine("Connect to TDengine success");}return conn;}}}// output:// Connect to TDengine success// fieldCount=6// ts current voltage phase location groupid// 1648432611249 10.3 219 0.31 California.SanFrancisco 2// 1648432611749 12.6 218 0.33 California.SanFrancisco 2
// compile with:// gcc -o query_example query_example.c -ltaos#include <inttypes.h>#include <stdio.h>#include <stdlib.h>#include <string.h>#include <taos.h>typedef int16_t VarDataLenT;#define TSDB_NCHAR_SIZE sizeof(int32_t)#define VARSTR_HEADER_SIZE sizeof(VarDataLenT)#define GET_FLOAT_VAL(x) (*(float *)(x))#define GET_DOUBLE_VAL(x) (*(double *)(x))#define varDataLen(v) ((VarDataLenT *)(v))[0]int printRow(char *str, TAOS_ROW row, TAOS_FIELD *fields, int numFields) {int len = 0;char split = ' ';for (int i = 0; i < numFields; ++i) {if (i > 0) {str[len++] = split;}if (row[i] == NULL) {len += sprintf(str + len, "%s", "NULL");continue;}switch (fields[i].type) {case TSDB_DATA_TYPE_TINYINT:len += sprintf(str + len, "%d", *((int8_t *)row[i]));break;case TSDB_DATA_TYPE_UTINYINT:len += sprintf(str + len, "%u", *((uint8_t *)row[i]));break;case TSDB_DATA_TYPE_SMALLINT:len += sprintf(str + len, "%d", *((int16_t *)row[i]));break;case TSDB_DATA_TYPE_USMALLINT:len += sprintf(str + len, "%u", *((uint16_t *)row[i]));break;case TSDB_DATA_TYPE_INT:len += sprintf(str + len, "%d", *((int32_t *)row[i]));break;case TSDB_DATA_TYPE_UINT:len += sprintf(str + len, "%u", *((uint32_t *)row[i]));break;case TSDB_DATA_TYPE_BIGINT:len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));break;case TSDB_DATA_TYPE_UBIGINT:len += sprintf(str + len, "%" PRIu64, *((uint64_t *)row[i]));break;case TSDB_DATA_TYPE_FLOAT: {float fv = 0;fv = GET_FLOAT_VAL(row[i]);len += sprintf(str + len, "%f", fv);} break;case TSDB_DATA_TYPE_DOUBLE: {double dv = 0;dv = GET_DOUBLE_VAL(row[i]);len += sprintf(str + len, "%lf", dv);} break;case TSDB_DATA_TYPE_BINARY:case TSDB_DATA_TYPE_NCHAR: {int32_t charLen = varDataLen((char *)row[i] - VARSTR_HEADER_SIZE);memcpy(str + len, row[i], charLen);len += charLen;} break;case TSDB_DATA_TYPE_TIMESTAMP:len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));break;case TSDB_DATA_TYPE_BOOL:len += sprintf(str + len, "%d", *((int8_t *)row[i]));default:break;}}return len;}/*** @brief print column name and values of each row** @param res* @return int*/static int printResult(TAOS_RES *res) {int numFields = taos_num_fields(res);TAOS_FIELD *fields = taos_fetch_fields(res);char header[256] = {0};int len = 0;for (int i = 0; i < numFields; ++i) {len += sprintf(header + len, "%s ", fields[i].name);}puts(header);TAOS_ROW row = NULL;while ((row = taos_fetch_row(res))) {char temp[256] = {0};printRow(temp, row, fields, numFields);puts(temp);}}int main() {TAOS *taos = taos_connect("localhost", "root", "taosdata", "power", 6030);if (taos == NULL) {puts("failed to connect to server");exit(EXIT_FAILURE);}TAOS_RES *res = taos_query(taos, "SELECT * FROM meters LIMIT 2");if (taos_errno(res) != 0) {printf("failed to execute taos_query. error: %s\n", taos_errstr(res));exit(EXIT_FAILURE);}printResult(res);taos_free_result(res);taos_close(taos);taos_cleanup();}// output:// ts current voltage phase location groupid// 1648432611249 10.300000 219 0.310000 California.SanFrancisco 2// 1648432611749 12.600000 218 0.330000 California.SanFrancisco 2
<?phpuse TDengine\Connection;use TDengine\Exception\TDengineException;try {// instantiate$host = 'localhost';$port = 6030;$username = 'root';$password = 'taosdata';$dbname = 'power';$connection = new Connection($host, $port, $username, $password, $dbname);// connect$connection->connect();$resource = $connection->query('SELECT ts, current FROM meters LIMIT 2');var_dump($resource->fetch());} catch (TDengineException $e) {// throw exceptionthrow $e;}
note
- With either REST connection or native connection, the above sample code works well.
- Please note that
use dbcan’t be used in case of REST connection because it’s stateless.
Asynchronous Query
Besides synchronous queries, an asynchronous query API is also provided by TDengine to insert or query data more efficiently. With a similar hardware and software environment, the async API is 2~4 times faster than sync APIs. Async API works in non-blocking mode, which means an operation can be returned without finishing so that the calling thread can switch to other work to improve the performance of the whole application system. Async APIs perform especially better in the case of poor networks.
Please note that async query can only be used with a native connection.
- Python
- C#
- C
import timefrom ctypes import *from taos import *def fetch_callback(p_param, p_result, num_of_rows):print("fetched ", num_of_rows, "rows")p = cast(p_param, POINTER(Counter))result = TaosResult(p_result)if num_of_rows == 0:print("fetching completed")p.contents.done = Trueresult.close()returnif num_of_rows < 0:p.contents.done = Trueresult.check_error(num_of_rows)result.close()return Nonefor row in result.rows_iter(num_of_rows):print(row)p.contents.count += result.row_countresult.fetch_rows_a(fetch_callback, p_param)def query_callback(p_param, p_result, code):if p_result is None:returnresult = TaosResult(p_result)if code == 0:result.fetch_rows_a(fetch_callback, p_param)result.check_error(code)class Counter(Structure):_fields_ = [("count", c_int), ("done", c_bool)]def __str__(self):return "{ count: %d, done: %s }" % (self.count, self.done)def test_query(conn):counter = Counter(count=0)conn.query_a("select ts, current, voltage from power.meters", query_callback, byref(counter))while not counter.done:print(counter)time.sleep(1)print(counter)conn.close()if __name__ == "__main__":test_query(connect())# possible output:# { count: 0, done: False }# fetched 8 rows# 1538548685000 10.300000 219# 1538548695000 12.600000 218# 1538548696800 12.300000 221# 1538548696650 10.300000 218# 1538548685500 11.800000 221# 1538548696600 13.400000 223# 1538548685500 10.800000 223# 1538548686500 11.500000 221# fetched 0 rows# fetching completed# { count: 8, done: True }
note
This sample code can’t be run on Windows system for now.
using System;using System.Collections.Generic;using TDengineDriver;using TDengineDriver.Impl;using System.Runtime.InteropServices;namespace TDengineExample{public class AsyncQueryExample{static void Main(){IntPtr conn = GetConnection();try{QueryAsyncCallback queryAsyncCallback = new QueryAsyncCallback(QueryCallback);TDengine.QueryAsync(conn, "select * from meters", queryAsyncCallback, IntPtr.Zero);Thread.Sleep(2000);}finally{TDengine.Close(conn);}}static void QueryCallback(IntPtr param, IntPtr taosRes, int code){if (code == 0 && taosRes != IntPtr.Zero){FetchRawBlockAsyncCallback fetchRowAsyncCallback = new FetchRawBlockAsyncCallback(FetchRawBlockCallback);TDengine.FetchRawBlockAsync(taosRes, fetchRowAsyncCallback, param);}else{throw new Exception($"async query data failed,code:{code},reason:{TDengine.Error(taosRes)}");}}// Iteratively call this interface until "numOfRows" is no greater than 0.static void FetchRawBlockCallback(IntPtr param, IntPtr taosRes, int numOfRows){if (numOfRows > 0){Console.WriteLine($"{numOfRows} rows async retrieved");IntPtr pdata = TDengine.GetRawBlock(taosRes);List<TDengineMeta> metaList = TDengine.FetchFields(taosRes);List<object> dataList = LibTaos.ReadRawBlock(pdata, metaList, numOfRows);for (int i = 0; i < dataList.Count; i++){if (i != 0 && (i + 1) % metaList.Count == 0){Console.WriteLine("{0}\t|", dataList[i]);}else{Console.Write("{0}\t|", dataList[i]);}}Console.WriteLine("");TDengine.FetchRawBlockAsync(taosRes, FetchRawBlockCallback, param);}else{if (numOfRows == 0){Console.WriteLine("async retrieve complete.");}else{throw new Exception($"FetchRawBlockCallback callback error, error code {numOfRows}");}TDengine.FreeResult(taosRes);}}static IntPtr GetConnection(){string host = "localhost";short port = 6030;string username = "root";string password = "taosdata";string dbname = "power";var conn = TDengine.Connect(host, username, password, dbname, port);if (conn == IntPtr.Zero){throw new Exception("Connect to TDengine failed");}else{Console.WriteLine("Connect to TDengine success");}return conn;}}}// //output:// // Connect to TDengine success// // 8 rows async retrieved// // 1538548685500 | 11.8 | 221 | 0.28 | california.losangeles | 2 |// // 1538548696600 | 13.4 | 223 | 0.29 | california.losangeles | 2 |// // 1538548685000 | 10.8 | 223 | 0.29 | california.losangeles | 3 |// // 1538548686500 | 11.5 | 221 | 0.35 | california.losangeles | 3 |// // 1538548685000 | 10.3 | 219 | 0.31 | california.sanfrancisco | 2 |// // 1538548695000 | 12.6 | 218 | 0.33 | california.sanfrancisco | 2 |// // 1538548696800 | 12.3 | 221 | 0.31 | california.sanfrancisco | 2 |// // 1538548696650 | 10.3 | 218 | 0.25 | california.sanfrancisco | 3 |// // async retrieve complete.
/*** @brief call back function of taos_fetch_row_a** @param param : the third parameter you passed to taos_fetch_row_a* @param res : pointer of TAOS_RES* @param numOfRow : number of rows fetched in this batch. will be 0 if there is no more data.* @return void**/void *fetch_row_callback(void *param, TAOS_RES *res, int numOfRow) {printf("numOfRow = %d \n", numOfRow);int numFields = taos_num_fields(res);TAOS_FIELD *fields = taos_fetch_fields(res);TAOS *_taos = (TAOS *)param;if (numOfRow > 0) {for (int i = 0; i < numOfRow; ++i) {TAOS_ROW row = taos_fetch_row(res);char temp[256] = {0};printRow(temp, row, fields, numFields);puts(temp);}taos_fetch_rows_a(res, fetch_row_callback, _taos);} else {printf("no more data, close the connection.\n");taos_free_result(res);taos_close(_taos);taos_cleanup();}}/*** @brief callback function of taos_query_a** @param param: the fourth parameter you passed to taos_query_a* @param res : the result set* @param code : status code* @return void**/void *select_callback(void *param, TAOS_RES *res, int code) {printf("query callback ...\n");TAOS *_taos = (TAOS *)param;if (code == 0 && res) {printHeader(res);taos_fetch_rows_a(res, fetch_row_callback, _taos);} else {printf("failed to execute taos_query. error: %s\n", taos_errstr(res));taos_free_result(res);taos_close(_taos);taos_cleanup();exit(EXIT_FAILURE);}}int main() {TAOS *taos = taos_connect("localhost", "root", "taosdata", "power", 6030);if (taos == NULL) {puts("failed to connect to server");exit(EXIT_FAILURE);}// param one is the connection returned by taos_connect.// param two is the SQL to execute.// param three is the callback function.// param four can be any pointer. It will be passed to your callback function as the first parameter. we use taos// here, because we want to close it after getting data.taos_query_a(taos, "SELECT * FROM meters", select_callback, taos);sleep(1);}// output:// query callback ...// ts current voltage phase location groupid// numOfRow = 8// 1538548685500 11.800000 221 0.280000 california.losangeles 2// 1538548696600 13.400000 223 0.290000 california.losangeles 2// 1538548685000 10.800000 223 0.290000 california.losangeles 3// 1538548686500 11.500000 221 0.350000 california.losangeles 3// 1538548685000 10.300000 219 0.310000 california.sanfrancisco 2// 1538548695000 12.600000 218 0.330000 california.sanfrancisco 2// 1538548696800 12.300000 221 0.310000 california.sanfrancisco 2// 1538548696650 10.300000 218 0.250000 california.sanfrancisco 3// numOfRow = 0// no more data, close the connection.
