This is an automated email from the ASF dual-hosted git repository.
CritasWang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
The following commit(s) were added to refs/heads/master by this push:
new 85c87a4 Add table-model query support to the Grafana data source
plugin (#114)
85c87a4 is described below
commit 85c87a4d5e3747541808b5759d50602e41c44809
Author: ZIHAN DAI <[email protected]>
AuthorDate: Thu Jul 23 16:17:33 2026 +1000
Add table-model query support to the Grafana data source plugin (#114)
* Add table-model query support to the Grafana data source plugin
The Grafana data source plugin only understood the tree model: both
QueryEditor modes build root.* paths and the backend calls
/grafana/v1/query/expression, which has no table-model equivalent, so
table-model users could not visualize their data through the plugin.
This adds a third QueryEditor mode, "SQL: Table Model", that runs standard
SQL against a database and renders the result. It reuses IoTDB's existing
table REST endpoint (POST /rest/table/v1/query) rather than adding a new
server API: the backend sends {database, sql} and turns the row-major
QueryDataSet into a Grafana frame with one typed, nullable field per
column, driven by the response's data_types so column types are exact
instead of guessed. The body is decoded with UseNumber so INT64 values
beyond 2^53 keep their precision.
In the default Time series format, rows are sorted by the first TIMESTAMP
column and a long-shaped result (time + tag columns + value columns) is
pivoted into one labeled series per tag combination, so a multi-device
query renders as separate lines; the Table format preserves the server's
row order. A $__timeFilter(col) / $__timeFrom / $__timeTo macro set
expands to the panel's range, accepting the $__timeFrom() / $__timeTo()
function forms as well. Because the table REST endpoint returns TIMESTAMP
values (and compares integer time literals) in the server's configured
timestamp_precision -- unlike the tree-model /grafana/v1 endpoints, which
normalize to milliseconds server-side -- a new data source "time
precision" option (ms default / us / ns) scales both the macro bounds and
the TIMESTAMP rendering, so the mode also works on us/ns servers.
Follow-up work for the same issue: a database -> table -> column browser,
a visual query builder, a default row limit to bound large SELECTs, and
the Grafana plugin SDK / Go backend modernization.
Signed-off-by: Zihan Dai <[email protected]>
* Switch the table-model backend from REST to the native Go client
Per review: query table-model data through apache/iotdb-client-go
(v2.0.8, TableSession) on the RPC port instead of POST
/rest/table/v1/query.
- A lazily created, per-datasource TableSessionPool serves the queries;
a new "rpc address" datasource option sets the endpoint, defaulting to
the URL's host with the default RPC port 6667. Tree-model queries and
the health check keep using the REST service for now.
- Values arrive as typed Go values instead of JSON, removing the
UseNumber/json.Number coercion layer; INT64 precision is inherent, and
the client converts TIMESTAMP/DATE values with the server-reported
timestamp precision, so the short-lived "time precision" datasource
option is removed again.
- The time macros now expand to ISO 8601 UTC timestamp literals
(e.g. 2020-09-13T12:26:40.000+00:00) rather than epoch-ms integers,
which the server parses correctly under any configured
timestamp_precision.
- The per-query database is applied with USE on the pooled session;
DATE renders as yyyy-MM-dd and BLOB as 0x-prefixed hex, matching the
REST transport's output.
- The server-side REST row cap (rest_query_default_row_size_limit) no
longer applies; the README now recommends a LIMIT clause for wide
scans.
- Frame building, the Time series/Table formats, the long-to-wide pivot
and the row sorting are unchanged; the fetch path sits behind a narrow
result-set interface so it stays unit-testable. 13 unit tests, go vet
and tsc --noEmit pass.
Signed-off-by: Zihan Dai <[email protected]>
* Require the database field in table-model mode
Per review: queryTableModel runs USE <database> on a pooled session, and
PooledTableSession.Close() only restores the database when the pool has
one configured — which this pool does not — so a session's USE state
survived release. A query with an empty database field could then
nondeterministically inherit whichever database a previous panel had
selected on that session.
Make the database required for the table-model mode: verifyQuery rejects
an empty database, the editor marks the field required, and the README
documents it. Every query therefore always runs USE for its own database
on the session it checked out, so leftover session state can never
influence a result regardless of pool scheduling. Fully-qualified
database.table references keep working and take precedence over the
session database.
Verified against a live IoTDB 2.0.8: alternating panels on two databases
sharing one session pool each saw exactly their own data across
repeated rounds.
Signed-off-by: Zihan Dai <[email protected]>
---------
Signed-off-by: Zihan Dai <[email protected]>
---
connectors/grafana-plugin/README.md | 33 +-
connectors/grafana-plugin/go.mod | 6 +-
connectors/grafana-plugin/pkg/plugin/plugin.go | 40 +-
.../grafana-plugin/pkg/plugin/table_query.go | 474 +++++++++++++++++++++
.../grafana-plugin/pkg/plugin/table_query_test.go | 422 ++++++++++++++++++
connectors/grafana-plugin/src/ConfigEditor.tsx | 21 +
connectors/grafana-plugin/src/QueryEditor.tsx | 93 +++-
connectors/grafana-plugin/src/datasource.ts | 7 +
connectors/grafana-plugin/src/types.ts | 11 +
9 files changed, 1098 insertions(+), 9 deletions(-)
diff --git a/connectors/grafana-plugin/README.md
b/connectors/grafana-plugin/README.md
index 974916b..b380501 100644
--- a/connectors/grafana-plugin/README.md
+++ b/connectors/grafana-plugin/README.md
@@ -69,7 +69,7 @@ Click the `New Dashboard` icon on the top right, and select
`Add an empty panel`
<img style="width:100%; max-width:800px; max-height:600px; margin-left:auto;
margin-right:auto; display:block;"
src="https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/add%20empty%20panel.png?raw=true";>
-Grafana plugin supports SQL: Full Customized mode and SQL: Drop-down List
mode, and the default mode is SQL: Full Customized mode.
+Grafana plugin supports SQL: Full Customized mode and SQL: Drop-down List mode
for the tree model, and SQL: Table Model mode for the table model. The default
mode is SQL: Full Customized mode.
<img style="width:100%; max-width:800px; max-height:600px; margin-left:auto;
margin-right:auto; display:block;"
src="https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/grafana_input_style.png?raw=true";>
@@ -119,6 +119,37 @@ Select a time series in the TIME-SERIES selection box,
select a function in the
<img style="width:100%; max-width:800px; max-height:600px; margin-left:auto;
margin-right:auto; display:block;"
src="https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/grafana_input2.png?raw=true";>
+##### SQL: Table Model
+
+Queries IoTDB's table model (IoTDB 2.x) through the native Go client
([apache/iotdb-client-go](https://github.com/apache/iotdb-client-go)). Enter a
standard SQL statement in the SQL input box, for example:
+
+```sql
+SELECT time, device_id, temperature FROM table1 WHERE $__timeFilter(time)
+```
+
+Unlike the tree-model modes, which go through the REST service, this mode
connects to the IoTDB RPC port (6667 by default). The data source's `rpc
address` option sets that endpoint explicitly; when it is left empty, the URL's
host with port 6667 is used.
+
+DATABASE input box: the database to run the statement against. Required —
every query first runs `USE <database>` on its pooled session, so results stay
deterministic regardless of which session the pool hands out. Fully-qualified
`database.table` references are still allowed and take precedence over the
session database.
+
+SQL input box: a single table-model SELECT statement. The following macros are
expanded by the plugin before the statement is sent:
+
+| Macro | Expands to |
+| ----- | ---------- |
+| `$__timeFilter(col)` | `(col >= <panel start> AND col <= <panel end>)`;
`col` defaults to `time` when omitted |
+| `$__timeFrom`, `$__timeFrom()` | the panel range start as an ISO 8601 UTC
timestamp literal |
+| `$__timeTo`, `$__timeTo()` | the panel range end as an ISO 8601 UTC
timestamp literal |
+
+The macros expand to ISO 8601 timestamp literals (e.g.
`2020-09-13T12:26:40.000+00:00`), which the server interprets in its own
configured `timestamp_precision` — so time filtering works unchanged on `ms`,
`us` and `ns` servers, and TIMESTAMP values are likewise converted by the
client using the server-reported precision.
+
+FORMAT option:
+
+* `Time series` (default): rows are sorted by the first TIMESTAMP column, and
a result that contains tag/string columns next to numeric columns is pivoted
into one series per tag combination — so a query returning several devices
renders as separate lines in a time-series panel. Note that the pivot requires
timestamps without nulls; results are sorted by the plugin, so an `ORDER BY`
clause is not needed.
+* `Table`: rows are returned exactly as the server sent them (use this with
the table panel, or when your `ORDER BY` should be preserved).
+
+Result columns are typed from the result set's data types: TIMESTAMP columns
become Grafana time fields, INT32/INT64 become integers, FLOAT/DOUBLE become
floats, BOOLEAN becomes booleans, and TEXT/STRING/DATE/BLOB render as strings.
+
+There is no server-imposed row cap on this path; add a `LIMIT` clause when
querying wide time ranges over dense data to bound the result size.
+
#### Support for variables and template functions
Both SQL: Full Customized and SQL: Drop-down List input methods support the
variable and template functions of grafana. In the following example, raw input
method is used, and aggregation is similar.
diff --git a/connectors/grafana-plugin/go.mod b/connectors/grafana-plugin/go.mod
index 8f7ca6b..fbc0951 100644
--- a/connectors/grafana-plugin/go.mod
+++ b/connectors/grafana-plugin/go.mod
@@ -19,11 +19,15 @@ go 1.21
toolchain go1.21.0
-require github.com/grafana/grafana-plugin-sdk-go v0.250.0
+require (
+ github.com/apache/iotdb-client-go/v2 v2.0.8
+ github.com/grafana/grafana-plugin-sdk-go v0.250.0
+)
require (
github.com/BurntSushi/toml v1.3.2 // indirect
github.com/apache/arrow/go/v15 v15.0.2 // indirect
+ github.com/apache/thrift v0.17.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go
b/connectors/grafana-plugin/pkg/plugin/plugin.go
index a65e987..4b362ba 100644
--- a/connectors/grafana-plugin/pkg/plugin/plugin.go
+++ b/connectors/grafana-plugin/pkg/plugin/plugin.go
@@ -28,8 +28,10 @@ import (
"net/http"
"strconv"
"strings"
+ "sync"
"time"
+ "github.com/apache/iotdb-client-go/v2/client"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
@@ -70,7 +72,8 @@ func ApacheIoTDBDatasource(ctx context.Context, d
backend.DataSourceInstanceSett
if password, exists := d.DecryptedSecureJSONData["password"]; exists {
authorization = "Basic " +
base64.StdEncoding.EncodeToString([]byte(dm.Username+":"+password))
}
- return &IoTDBDataSource{CallResourceHandler:
iotdbResourceHandler(authorization, httpClient), Username: dm.Username, Ulr:
dm.Url, httpClient: httpClient}, nil
+ password := d.DecryptedSecureJSONData["password"]
+ return &IoTDBDataSource{CallResourceHandler:
iotdbResourceHandler(authorization, httpClient), Username: dm.Username, Ulr:
dm.Url, RPCAddress: dm.RPCAddress, password: password, httpClient: httpClient},
nil
}
// SampleDatasource is an example datasource which can respond to data
queries, reports
@@ -79,7 +82,14 @@ type IoTDBDataSource struct {
backend.CallResourceHandler
Username string
Ulr string
+ RPCAddress string
+ password string
httpClient *http.Client
+
+ // Native-client session pool for the table-model mode, created lazily
by
+ // getTablePool on the first table query.
+ tablePoolMu sync.Mutex
+ tablePool *client.TableSessionPool
}
// Dispose here tells plugin SDK that plugin wants to clean up resources when
a new instance
@@ -88,6 +98,12 @@ type IoTDBDataSource struct {
func (d *IoTDBDataSource) Dispose() {
// Clean up datasource instance resources.
d.httpClient.CloseIdleConnections()
+ d.tablePoolMu.Lock()
+ defer d.tablePoolMu.Unlock()
+ if d.tablePool != nil {
+ d.tablePool.Close()
+ d.tablePool = nil
+ }
}
// QueryData handles multiple queries and returns multiple responses.
@@ -113,6 +129,10 @@ func (d *IoTDBDataSource) QueryData(ctx context.Context,
req *backend.QueryDataR
type dataSourceModel struct {
Username string `json:"username"`
Url string `json:"url"`
+ // RPCAddress is the IoTDB RPC endpoint (host or host:port) the native
+ // client used by the table-model mode connects to. When empty, the
URL's
+ // host with the default RPC port 6667 is used.
+ RPCAddress string `json:"rpcAddress"`
}
type groupBy struct {
@@ -134,6 +154,9 @@ type queryParam struct {
FillClauses string `json:"fillClauses"`
GroupBy groupBy `json:"groupBy"`
Hide bool `json:"hide"`
+ Database string `json:"database"`
+ Sql string `json:"sql"`
+ Format string `json:"format"`
}
type QueryDataReq struct {
@@ -200,6 +223,17 @@ func verifyQuery(query backend.DataQuery) (qp *queryParam,
errMsg string) {
return nil, "Input error, FROM is required"
}
}
+ } else if qp.SqlType == TableModelSqlType {
+ if strings.TrimSpace(qp.Sql) == "" {
+ return nil, "Input error, SQL is required"
+ }
+ // The database is required: every query then USEs it on its
pooled
+ // session, so a session's leftover USE state from a previous
query
+ // can never influence the result (PooledTableSession.Close only
+ // resets the database when the pool has one configured).
+ if strings.TrimSpace(qp.Database) == "" {
+ return nil, "Input error, DATABASE is required"
+ }
} else {
return nil, "none"
}
@@ -231,6 +265,10 @@ func (d *IoTDBDataSource) query(cxt context.Context, pCtx
backend.PluginContext,
qp.StartTime = query.TimeRange.From.UnixNano() / 1000000
qp.EndTime = query.TimeRange.To.UnixNano() / 1000000
+ if qp.SqlType == TableModelSqlType {
+ return d.queryTableModel(cxt, qp)
+ }
+
if qp.SqlType == "SQL: Drop-down List" {
qp.Control = ""
var expressions []string = qp.Paths[len(qp.Paths)-1:]
diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go
b/connectors/grafana-plugin/pkg/plugin/table_query.go
new file mode 100644
index 0000000..a53eff1
--- /dev/null
+++ b/connectors/grafana-plugin/pkg/plugin/table_query.go
@@ -0,0 +1,474 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package plugin
+
+import (
+ "context"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "net"
+ "net/url"
+ "regexp"
+ "sort"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/apache/iotdb-client-go/v2/client"
+ "github.com/grafana/grafana-plugin-sdk-go/backend"
+ "github.com/grafana/grafana-plugin-sdk-go/backend/log"
+ "github.com/grafana/grafana-plugin-sdk-go/data"
+)
+
+// TableModelSqlType is the QueryEditor mode that runs standard table-model SQL
+// through the native Go client (apache/iotdb-client-go), as opposed to the
+// tree-model modes ("SQL: Full Customized" / "SQL: Drop-down List") that build
+// root.* paths and go through the REST /grafana endpoints.
+const TableModelSqlType = "SQL: Table Model"
+
+// Result formats for the table-model mode. Time series (the default) sorts
+// rows ascending by the first TIMESTAMP column and pivots a long-shaped result
+// (time + tag columns + value columns) into one labeled series per tag
+// combination; Table returns the rows exactly as the server sent them.
+const (
+ tableFormatTimeSeries = "Time series"
+ tableFormatTable = "Table"
+)
+
+// Connection settings for the native client. The RPC endpoint is the
+// datasource's "rpc address" option, or the URL's host with the default RPC
+// port when unset. The pool is created lazily on the first table-model query
+// so datasources that only use the tree model never open an RPC connection.
+const (
+ defaultRPCPort = "6667"
+ tablePoolMaxSize = 8
+ tableConnectTimeoutMs = 10000
+ tableWaitSessionTimeoutMs = 60000
+ defaultQueryTimeoutMs = int64(60000)
+)
+
+// timeFilterRe matches Grafana's $__timeFilter(column) macro; the column is
+// optional and defaults to "time". One level of nested parentheses is allowed
+// so expressions like $__timeFilter(cast(x)) survive intact.
+var timeFilterRe =
regexp.MustCompile(`\$__timeFilter\(\s*((?:[^()]|\([^()]*\))*?)\s*\)`)
+
+// timeFromRe / timeToRe match $__timeFrom / $__timeTo, with or without the
+// trailing () that Grafana's SQL data sources use ($__timeFrom()).
+var (
+ timeFromRe = regexp.MustCompile(`\$__timeFrom\b(?:\s*\(\s*\))?`)
+ timeToRe = regexp.MustCompile(`\$__timeTo\b(?:\s*\(\s*\))?`)
+)
+
+// formatTimeLiteral renders a panel-range bound as an ISO 8601 UTC timestamp
+// literal (e.g. 2020-09-13T12:26:40.000+00:00). The server parses such a
+// literal in its own configured timestamp precision, so the expansion works
+// unchanged on ms, us and ns servers — unlike a bare epoch integer, which the
+// server would interpret in raw server units.
+func formatTimeLiteral(ms int64) string {
+ return time.UnixMilli(ms).UTC().Format("2006-01-02T15:04:05.000") +
"+00:00"
+}
+
+// expandTableMacros rewrites the Grafana time macros a dashboard author can
put
+// in table-model SQL into concrete bounds for the panel's range:
+//
+// $__timeFilter(col) -> (col >= <from> AND col <= <to>)
+// $__timeFrom[()] -> <from>
+// $__timeTo[()] -> <to>
+//
+// Bounds are ISO 8601 UTC timestamp literals, which IoTDB compares against
+// TIMESTAMP columns independently of the server's timestamp precision.
+func expandTableMacros(sql string, startMs int64, endMs int64) string {
+ from := formatTimeLiteral(startMs)
+ to := formatTimeLiteral(endMs)
+ sql = timeFilterRe.ReplaceAllStringFunc(sql, func(m string) string {
+ col := strings.TrimSpace(timeFilterRe.FindStringSubmatch(m)[1])
+ if col == "" {
+ col = "time"
+ }
+ return "(" + col + " >= " + from + " AND " + col + " <= " + to
+ ")"
+ })
+ sql = timeFromRe.ReplaceAllString(sql, from)
+ sql = timeToRe.ReplaceAllString(sql, to)
+ return sql
+}
+
+// quoteTableIdentifier wraps a table-model identifier in double quotes
+// (doubling any embedded quote), the relational grammar's quoted-identifier
+// form, so a database name survives the USE statement verbatim.
+func quoteTableIdentifier(name string) string {
+ return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
+}
+
+// tableQueryDataSet is the plugin-internal carrier for a fetched table-model
+// result: column names and IoTDB type names as reported by the client, plus
+// row-major cell values (values[row][col]) holding the client's native Go
+// representations (time.Time for TIMESTAMP/DATE, string for TEXT/STRING,
+// []byte for BLOB, bool/int32/int64/float32/float64 for the scalars, nil for
+// null).
+type tableQueryDataSet struct {
+ ColumnNames []string
+ DataTypes []string
+ Values [][]interface{}
+}
+
+// tableResultSet is the slice of the native client's SessionDataSet the fetch
+// path needs; narrowing it to an interface keeps fetchTableDataSet testable
+// without a live server. Column indexes are 1-based, matching the client.
+type tableResultSet interface {
+ Next() (bool, error)
+ GetColumnNames() []string
+ GetColumnTypes() []string
+ GetObjectByIndex(columnIndex int32) (interface{}, error)
+ Close() error
+}
+
+// fetchTableDataSet drains a result set into a tableQueryDataSet. The caller
+// owns closing the result set.
+func fetchTableDataSet(rs tableResultSet) (*tableQueryDataSet, error) {
+ names := rs.GetColumnNames()
+ types := rs.GetColumnTypes()
+ dataSet := &tableQueryDataSet{ColumnNames: names, DataTypes: types,
Values: [][]interface{}{}}
+ for {
+ hasNext, err := rs.Next()
+ if err != nil {
+ return nil, err
+ }
+ if !hasNext {
+ return dataSet, nil
+ }
+ row := make([]interface{}, len(names))
+ for i := range names {
+ value, err := rs.GetObjectByIndex(int32(i) + 1)
+ if err != nil {
+ return nil, err
+ }
+ row[i] = value
+ }
+ dataSet.Values = append(dataSet.Values, row)
+ }
+}
+
+// tableRPCEndpoint resolves the host and port of the IoTDB RPC service the
+// native client connects to: the datasource's "rpc address" option when set
+// (host or host:port), otherwise the datasource URL's host with the default
+// RPC port.
+func (d *IoTDBDataSource) tableRPCEndpoint() (string, string, error) {
+ if addr := strings.TrimSpace(d.RPCAddress); addr != "" {
+ if host, port, err := net.SplitHostPort(addr); err == nil {
+ return host, port, nil
+ }
+ return strings.Trim(addr, "[]"), defaultRPCPort, nil
+ }
+ raw := strings.TrimSpace(d.Ulr)
+ if u, err := url.Parse(raw); err == nil && u.Hostname() != "" {
+ return u.Hostname(), defaultRPCPort, nil
+ }
+ // A bare host[:restPort] without a scheme parses as opaque; retry with
one.
+ if u, err := url.Parse("http://"; + raw); err == nil && u.Hostname() !=
"" {
+ return u.Hostname(), defaultRPCPort, nil
+ }
+ return "", "", errors.New("cannot derive the IoTDB RPC host from the
datasource URL; please set the rpc address option")
+}
+
+// getTablePool lazily creates the shared native-client session pool for this
+// datasource instance. Pool construction does not connect; connection errors
+// surface on GetSession.
+func (d *IoTDBDataSource) getTablePool() (*client.TableSessionPool, error) {
+ d.tablePoolMu.Lock()
+ defer d.tablePoolMu.Unlock()
+ if d.tablePool != nil {
+ return d.tablePool, nil
+ }
+ host, port, err := d.tableRPCEndpoint()
+ if err != nil {
+ return nil, err
+ }
+ poolConfig := &client.PoolConfig{
+ Host: host,
+ Port: port,
+ UserName: d.Username,
+ Password: d.password,
+ }
+ pool := client.NewTableSessionPool(poolConfig, tablePoolMaxSize,
tableConnectTimeoutMs, tableWaitSessionTimeoutMs, false)
+ d.tablePool = &pool
+ return d.tablePool, nil
+}
+
+// queryTableModel runs a table-model SQL query through the native client and
+// turns the result set into a Grafana data frame.
+func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam)
backend.DataResponse {
+ response := backend.DataResponse{}
+
+ sql := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime)
+
+ pool, err := d.getTablePool()
+ if err != nil {
+ response.Error = err
+ return response
+ }
+ session, err := pool.GetSession()
+ if err != nil {
+ response.Error = fmt.Errorf("cannot connect to the IoTDB RPC
service: %w", err)
+ log.DefaultLogger.Error("Cannot connect to the IoTDB RPC
service", "err", err)
+ return response
+ }
+ defer func() {
+ if closeErr := session.Close(); closeErr != nil {
+ log.DefaultLogger.Error("Failed to return session to
the pool", "err", closeErr)
+ }
+ }()
+
+ if database := strings.TrimSpace(qp.Database); database != "" {
+ if err := session.ExecuteNonQueryStatement("USE " +
quoteTableIdentifier(database)); err != nil {
+ response.Error = err
+ return response
+ }
+ }
+
+ timeout := defaultQueryTimeoutMs
+ if deadline, ok := ctx.Deadline(); ok {
+ if ms := time.Until(deadline).Milliseconds(); ms > 0 {
+ timeout = ms
+ }
+ }
+ resultSet, err := session.ExecuteQueryStatement(sql, &timeout)
+ if err != nil {
+ response.Error = err
+ return response
+ }
+ defer func() {
+ if closeErr := resultSet.Close(); closeErr != nil {
+ log.DefaultLogger.Error("Failed to close the result
set", "err", closeErr)
+ }
+ }()
+
+ dataSet, err := fetchTableDataSet(resultSet)
+ if err != nil {
+ response.Error = err
+ return response
+ }
+
+ response.Frames = append(response.Frames,
buildTableResponseFrame(dataSet, qp.Format))
+ return response
+}
+
+// buildTableResponseFrame turns a fetched dataset into the response frame,
+// honoring the query's format. In the default Time series format the rows are
+// sorted ascending by the first TIMESTAMP column and a long-shaped result
+// (time + string tag columns + value columns) is pivoted into one labeled
+// series per tag combination, so a multi-device query renders as separate
+// lines instead of one interleaved series; when the pivot does not apply (or
+// fails, e.g. on a null timestamp) the plain frame is returned. The Table
+// format preserves the server's row order untouched.
+func buildTableResponseFrame(dataSet *tableQueryDataSet, format string)
*data.Frame {
+ // Anything that is not explicitly the Table format gets the default
+ // time-series treatment, including queries saved before FORMAT existed.
+ isTimeSeries := !strings.EqualFold(format, tableFormatTable)
+ if isTimeSeries {
+ sortRowsByFirstTimestamp(dataSet)
+ }
+ frame := buildTableFrame(dataSet)
+ if isTimeSeries && frame.TimeSeriesSchema().Type ==
data.TimeSeriesTypeLong {
+ if wide, err := data.LongToWide(frame, nil); err == nil {
+ frame = wide
+ }
+ }
+ return frame
+}
+
+// sortRowsByFirstTimestamp stably sorts the row-major values ascending by the
+// first TIMESTAMP column, which both time-series rendering and the
long-to-wide
+// pivot require. Rows whose time cell is null sort last.
+func sortRowsByFirstTimestamp(dataSet *tableQueryDataSet) {
+ timeCol := -1
+ for i, t := range dataSet.DataTypes {
+ if strings.EqualFold(t, "TIMESTAMP") {
+ timeCol = i
+ break
+ }
+ }
+ if timeCol < 0 {
+ return
+ }
+ sort.SliceStable(dataSet.Values, func(a, b int) bool {
+ va, okA := rowTimeAt(dataSet.Values[a], timeCol)
+ vb, okB := rowTimeAt(dataSet.Values[b], timeCol)
+ if okA != okB {
+ return okA
+ }
+ return okA && va.Before(vb)
+ })
+}
+
+func rowTimeAt(row []interface{}, col int) (time.Time, bool) {
+ if col >= len(row) {
+ return time.Time{}, false
+ }
+ t, ok := row[col].(time.Time)
+ return t, ok
+}
+
+// buildTableFrame converts the row-major dataset into a Grafana frame, one
+// typed field per column driven by the reported data types; each field gathers
+// a single column across all rows, so every field ends up the same length (the
+// row count) that Grafana requires.
+func buildTableFrame(dataSet *tableQueryDataSet) *data.Frame {
+ frame := data.NewFrame("response")
+ rowCount := len(dataSet.Values)
+ for col := 0; col < len(dataSet.ColumnNames); col++ {
+ name := dataSet.ColumnNames[col]
+ dataType := ""
+ if col < len(dataSet.DataTypes) {
+ dataType = dataSet.DataTypes[col]
+ }
+ columnValues := make([]interface{}, rowCount)
+ for row := 0; row < rowCount; row++ {
+ if col < len(dataSet.Values[row]) {
+ columnValues[row] = dataSet.Values[row][col]
+ }
+ }
+ frame.Fields = append(frame.Fields, buildTableField(name,
dataType, columnValues))
+ }
+ return frame
+}
+
+// buildTableField turns one column's native client values into a typed,
+// nullable Grafana field selected by the IoTDB data type. The client already
+// converts TIMESTAMP (and DATE) values to time.Time using the server-reported
+// timestamp precision, so no unit handling happens here. DATE and BLOB render
+// as strings (yyyy-MM-dd and 0x-prefixed hex), matching the REST behavior the
+// mode previously had.
+func buildTableField(name string, dataType string, values []interface{})
*data.Field {
+ switch strings.ToUpper(dataType) {
+ case "TIMESTAMP":
+ out := make([]*time.Time, len(values))
+ for i, v := range values {
+ if t, ok := v.(time.Time); ok {
+ value := t
+ out[i] = &value
+ }
+ }
+ return data.NewField(name, nil, out)
+ case "INT32", "INT64":
+ out := make([]*int64, len(values))
+ for i, v := range values {
+ if n, ok := toInt64(v); ok {
+ value := n
+ out[i] = &value
+ }
+ }
+ return data.NewField(name, nil, out)
+ case "FLOAT", "DOUBLE":
+ out := make([]*float64, len(values))
+ for i, v := range values {
+ if f, ok := toFloat64(v); ok {
+ value := f
+ out[i] = &value
+ }
+ }
+ return data.NewField(name, nil, out)
+ case "BOOLEAN":
+ out := make([]*bool, len(values))
+ for i, v := range values {
+ if b, ok := v.(bool); ok {
+ value := b
+ out[i] = &value
+ }
+ }
+ return data.NewField(name, nil, out)
+ case "DATE":
+ out := make([]*string, len(values))
+ for i, v := range values {
+ if t, ok := v.(time.Time); ok {
+ value := t.Format("2006-01-02")
+ out[i] = &value
+ }
+ }
+ return data.NewField(name, nil, out)
+ case "BLOB":
+ out := make([]*string, len(values))
+ for i, v := range values {
+ if b, ok := v.([]byte); ok {
+ value := "0x" + hex.EncodeToString(b)
+ out[i] = &value
+ }
+ }
+ return data.NewField(name, nil, out)
+ default:
+ // TEXT, STRING and anything unrecognised render as strings.
+ out := make([]*string, len(values))
+ for i, v := range values {
+ if v == nil {
+ continue
+ }
+ value := toString(v)
+ out[i] = &value
+ }
+ return data.NewField(name, nil, out)
+ }
+}
+
+// toInt64 coerces the client's integer representations (INT32 -> int32,
+// INT64 -> int64) to int64.
+func toInt64(v interface{}) (int64, bool) {
+ switch n := v.(type) {
+ case int64:
+ return n, true
+ case int32:
+ return int64(n), true
+ }
+ return 0, false
+}
+
+// toFloat64 coerces the client's floating representations (FLOAT -> float32,
+// DOUBLE -> float64) to float64.
+func toFloat64(v interface{}) (float64, bool) {
+ switch n := v.(type) {
+ case float64:
+ return n, true
+ case float32:
+ return float64(n), true
+ }
+ return 0, false
+}
+
+// toString renders a value for a text column without losing it, whatever the
+// client handed over.
+func toString(v interface{}) string {
+ switch value := v.(type) {
+ case string:
+ return value
+ case []byte:
+ return "0x" + hex.EncodeToString(value)
+ case time.Time:
+ return value.UTC().Format(time.RFC3339Nano)
+ case bool:
+ return strconv.FormatBool(value)
+ case int32:
+ return strconv.FormatInt(int64(value), 10)
+ case int64:
+ return strconv.FormatInt(value, 10)
+ case float32:
+ return strconv.FormatFloat(float64(value), 'f', -1, 32)
+ case float64:
+ return strconv.FormatFloat(value, 'f', -1, 64)
+ default:
+ return fmt.Sprintf("%v", v)
+ }
+}
diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go
b/connectors/grafana-plugin/pkg/plugin/table_query_test.go
new file mode 100644
index 0000000..899bc14
--- /dev/null
+++ b/connectors/grafana-plugin/pkg/plugin/table_query_test.go
@@ -0,0 +1,422 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package plugin
+
+import (
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/grafana/grafana-plugin-sdk-go/backend"
+ "github.com/grafana/grafana-plugin-sdk-go/data"
+)
+
+func ts(ms int64) time.Time {
+ return time.UnixMilli(ms).UTC()
+}
+
+func TestExpandTableMacros(t *testing.T) {
+ const from int64 = 1600000000000 // 2020-09-13T12:26:40.000+00:00
+ const to int64 = 1600000001000 // 2020-09-13T12:26:41.000+00:00
+ const fromLit = "2020-09-13T12:26:40.000+00:00"
+ const toLit = "2020-09-13T12:26:41.000+00:00"
+
+ cases := []struct {
+ name string
+ in string
+ want string
+ }{
+ {
+ name: "timeFilter with explicit column",
+ in: "SELECT time, s0 FROM db.t WHERE
$__timeFilter(time)",
+ want: "SELECT time, s0 FROM db.t WHERE (time >= " +
fromLit + " AND time <= " + toLit + ")",
+ },
+ {
+ name: "timeFilter defaults to time column when empty",
+ in: "SELECT * FROM db.t WHERE $__timeFilter()",
+ want: "SELECT * FROM db.t WHERE (time >= " + fromLit +
" AND time <= " + toLit + ")",
+ },
+ {
+ name: "timeFrom and timeTo",
+ in: "SELECT * FROM db.t WHERE time >= $__timeFrom AND
time <= $__timeTo",
+ want: "SELECT * FROM db.t WHERE time >= " + fromLit + "
AND time <= " + toLit,
+ },
+ {
+ name: "function form timeFrom and timeTo",
+ in: "SELECT * FROM db.t WHERE time >= $__timeFrom()
AND time <= $__timeTo( )",
+ want: "SELECT * FROM db.t WHERE time >= " + fromLit + "
AND time <= " + toLit,
+ },
+ {
+ name: "timeFilter with one nested paren level",
+ in: "SELECT * FROM db.t WHERE $__timeFilter(cast(x))",
+ want: "SELECT * FROM db.t WHERE (cast(x) >= " + fromLit
+ " AND cast(x) <= " + toLit + ")",
+ },
+ {
+ name: "longer identifier is not mangled",
+ in: "SELECT $__timeFromage FROM db.t",
+ want: "SELECT $__timeFromage FROM db.t",
+ },
+ {
+ name: "no macros is unchanged",
+ in: "SELECT * FROM db.t",
+ want: "SELECT * FROM db.t",
+ },
+ }
+
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ got := expandTableMacros(c.in, from, to)
+ if got != c.want {
+ t.Fatalf("expandTableMacros() = %q, want %q",
got, c.want)
+ }
+ })
+ }
+}
+
+func TestQuoteTableIdentifier(t *testing.T) {
+ if got := quoteTableIdentifier("test"); got != `"test"` {
+ t.Fatalf("plain identifier = %q", got)
+ }
+ if got := quoteTableIdentifier(`we"ird`); got != `"we""ird"` {
+ t.Fatalf("embedded quote = %q", got)
+ }
+}
+
+// fakeResultSet implements tableResultSet over an in-memory row list, standing
+// in for the native client's SessionDataSet (1-based column indexes).
+type fakeResultSet struct {
+ names []string
+ types []string
+ rows [][]interface{}
+ cursor int
+ err error
+ closed bool
+}
+
+func (f *fakeResultSet) Next() (bool, error) {
+ if f.err != nil {
+ return false, f.err
+ }
+ if f.cursor >= len(f.rows) {
+ return false, nil
+ }
+ f.cursor++
+ return true, nil
+}
+
+func (f *fakeResultSet) GetColumnNames() []string { return f.names }
+func (f *fakeResultSet) GetColumnTypes() []string { return f.types }
+
+func (f *fakeResultSet) GetObjectByIndex(columnIndex int32) (interface{},
error) {
+ return f.rows[f.cursor-1][columnIndex-1], nil
+}
+
+func (f *fakeResultSet) Close() error {
+ f.closed = true
+ return nil
+}
+
+func TestFetchTableDataSet(t *testing.T) {
+ rs := &fakeResultSet{
+ names: []string{"time", "device", "value"},
+ types: []string{"TIMESTAMP", "STRING", "DOUBLE"},
+ rows: [][]interface{}{
+ {ts(1000), "d1", float64(1.5)},
+ {ts(2000), nil, float64(2.5)}, // null cell passes
through
+ },
+ }
+ dataSet, err := fetchTableDataSet(rs)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(dataSet.Values) != 2 || len(dataSet.ColumnNames) != 3 {
+ t.Fatalf("unexpected shape: %d rows, %d columns",
len(dataSet.Values), len(dataSet.ColumnNames))
+ }
+ if dataSet.Values[0][1] != "d1" || dataSet.Values[1][1] != nil {
+ t.Fatalf("cells not carried over: %#v", dataSet.Values)
+ }
+ if tv, ok := dataSet.Values[1][0].(time.Time); !ok ||
!tv.Equal(ts(2000)) {
+ t.Fatalf("time cell = %#v, want %v", dataSet.Values[1][0],
ts(2000))
+ }
+}
+
+func TestFetchTableDataSetPropagatesError(t *testing.T) {
+ rs := &fakeResultSet{names: []string{"a"}, types: []string{"INT64"},
err: errors.New("broken pipe")}
+ if _, err := fetchTableDataSet(rs); err == nil {
+ t.Fatalf("expected the iteration error to propagate")
+ }
+}
+
+// TestBuildTableFrameRowOrientation pins the fetch orientation contract: the
+// dataset rows are row-major (values[row][col]), so a field must gather a
+// single column across every row, with the client's native Go value types.
+func TestBuildTableFrameRowOrientation(t *testing.T) {
+ ds := &tableQueryDataSet{
+ ColumnNames: []string{"time", "i", "d", "b", "s"},
+ DataTypes: []string{"TIMESTAMP", "INT64", "DOUBLE",
"BOOLEAN", "TEXT"},
+ Values: [][]interface{}{
+ {ts(1600000000000), int64(42), float64(3.5), true,
"hello"},
+ {ts(1600000001000), int64(43), float64(4.5), false,
"world"},
+ },
+ }
+
+ frame := buildTableFrame(ds)
+ if len(frame.Fields) != 5 {
+ t.Fatalf("expected 5 fields, got %d", len(frame.Fields))
+ }
+ for _, f := range frame.Fields {
+ if f.Len() != 2 {
+ t.Fatalf("field %q has len %d, want 2 (row count)",
f.Name, f.Len())
+ }
+ }
+
+ if v, ok := frame.Fields[0].At(1).(*time.Time); !ok || v == nil ||
!v.Equal(ts(1600000001000)) {
+ t.Fatalf("time[1] = %#v", frame.Fields[0].At(1))
+ }
+ if v := frame.Fields[1].At(0).(*int64); *v != 42 {
+ t.Fatalf("i[0] = %d, want 42", *v)
+ }
+ if v := frame.Fields[2].At(1).(*float64); *v != 4.5 {
+ t.Fatalf("d[1] = %v, want 4.5", *v)
+ }
+ if v := frame.Fields[3].At(1).(*bool); *v != false {
+ t.Fatalf("b[1] = %v, want false", *v)
+ }
+ if v := frame.Fields[4].At(1).(*string); *v != "world" {
+ t.Fatalf("s[1] = %q, want world", *v)
+ }
+}
+
+// TestBuildTableFieldNativeTypeCoercions pins the client-type mapping: INT32
+// widens to int64, FLOAT widens to float64, an INT64 above 2^53 stays exact
+// (native int64, no float roundtrip), DATE renders as yyyy-MM-dd and BLOB as
+// 0x-prefixed hex — the same rendering the REST transport produced.
+func TestBuildTableFieldNativeTypeCoercions(t *testing.T) {
+ if v := buildTableField("i", "INT32",
[]interface{}{int32(7)}).At(0).(*int64); *v != 7 {
+ t.Fatalf("INT32 = %d, want 7", *v)
+ }
+ if v := buildTableField("f", "FLOAT",
[]interface{}{float32(1.5)}).At(0).(*float64); *v != 1.5 {
+ t.Fatalf("FLOAT = %v, want 1.5", *v)
+ }
+ if v := buildTableField("big", "INT64",
[]interface{}{int64(9007199254740993)}).At(0).(*int64); *v != 9007199254740993 {
+ t.Fatalf("INT64 precision lost: %d", *v)
+ }
+ date := time.Date(2025, 7, 14, 0, 0, 0, 0, time.UTC)
+ if v := buildTableField("e", "DATE",
[]interface{}{date}).At(0).(*string); *v != "2025-07-14" {
+ t.Fatalf("DATE = %q, want 2025-07-14", *v)
+ }
+ if v := buildTableField("d", "BLOB", []interface{}{[]byte{0xca, 0xfe,
0xba, 0xbe}}).At(0).(*string); *v != "0xcafebabe" {
+ t.Fatalf("BLOB = %q, want 0xcafebabe", *v)
+ }
+}
+
+func TestBuildTableFieldNullsBecomeNilPointers(t *testing.T) {
+ f := buildTableField("i", "INT64", []interface{}{int64(1), nil,
int64(3)})
+ if f.Len() != 3 {
+ t.Fatalf("expected 3 values, got %d", f.Len())
+ }
+ if v, ok := f.At(1).(*int64); !ok || v != nil {
+ t.Fatalf("null value should be a nil *int64, got %#v", f.At(1))
+ }
+ if v, ok := f.At(0).(*int64); !ok || v == nil || *v != 1 {
+ t.Fatalf("first value = %#v, want *int64(1)", f.At(0))
+ }
+}
+
+func TestBuildTableFieldUnknownTypeRendersString(t *testing.T) {
+ f := buildTableField("x", "SOMETHING_NEW", []interface{}{int64(7),
"raw", nil})
+ if f.Len() != 3 {
+ t.Fatalf("expected 3 values, got %d", f.Len())
+ }
+ if v, ok := f.At(0).(*string); !ok || v == nil || *v != "7" {
+ t.Fatalf("numeric coerced to string = %#v, want *string(7)",
f.At(0))
+ }
+ if v, ok := f.At(1).(*string); !ok || v == nil || *v != "raw" {
+ t.Fatalf("string value = %#v, want *string(raw)", f.At(1))
+ }
+ if v, ok := f.At(2).(*string); !ok || v != nil {
+ t.Fatalf("null value should be nil *string, got %#v", f.At(2))
+ }
+}
+
+// TestBuildTableFrameRaggedRowIsSafe checks that a row shorter than the column
+// header does not panic and yields equal-length, null-padded fields.
+func TestBuildTableFrameRaggedRowIsSafe(t *testing.T) {
+ ds := &tableQueryDataSet{
+ ColumnNames: []string{"time", "s0"},
+ DataTypes: []string{"TIMESTAMP", "TEXT"},
+ Values: [][]interface{}{
+ {ts(1), "a"},
+ {ts(2)}, // ragged: missing s0
+ },
+ }
+ frame := buildTableFrame(ds)
+ if len(frame.Fields) != 2 {
+ t.Fatalf("expected 2 fields, got %d", len(frame.Fields))
+ }
+ if frame.Fields[0].Len() != 2 || frame.Fields[1].Len() != 2 {
+ t.Fatalf("fields must be equal length (2), got %d and %d",
frame.Fields[0].Len(), frame.Fields[1].Len())
+ }
+ if v, ok := frame.Fields[1].At(1).(*string); !ok || v != nil {
+ t.Fatalf("missing ragged cell should be nil *string, got %#v",
frame.Fields[1].At(1))
+ }
+}
+
+// TestBuildTableResponseFrameLongToWide pins the multi-device path: a long
+// result (time + tag + value), even arriving unsorted, must come back as one
+// labeled series per tag value so a time-series panel draws separate lines.
+func TestBuildTableResponseFrameLongToWide(t *testing.T) {
+ ds := &tableQueryDataSet{
+ ColumnNames: []string{"time", "device", "temperature"},
+ DataTypes: []string{"TIMESTAMP", "STRING", "DOUBLE"},
+ Values: [][]interface{}{
+ {ts(2000), "d2", float64(22.5)},
+ {ts(1000), "d1", float64(11.0)},
+ {ts(1000), "d2", float64(21.0)},
+ {ts(2000), "d1", float64(12.0)},
+ },
+ }
+
+ frame := buildTableResponseFrame(ds, "")
+ if len(frame.Fields) != 3 {
+ t.Fatalf("expected time + one series per device (3 fields), got
%d", len(frame.Fields))
+ }
+ if frame.Fields[0].Len() != 2 {
+ t.Fatalf("expected 2 wide rows, got %d", frame.Fields[0].Len())
+ }
+ byDevice := map[string][]float64{}
+ for _, f := range frame.Fields[1:] {
+ dev := f.Labels["device"]
+ if dev == "" {
+ t.Fatalf("value field %q has no device label: %v",
f.Name, f.Labels)
+ }
+ var vals []float64
+ for i := 0; i < f.Len(); i++ {
+ v, ok := f.At(i).(*float64)
+ if !ok || v == nil {
+ t.Fatalf("field %q row %d = %#v, want
*float64", f.Name, i, f.At(i))
+ }
+ vals = append(vals, *v)
+ }
+ byDevice[dev] = vals
+ }
+ if v := byDevice["d1"]; len(v) != 2 || v[0] != 11.0 || v[1] != 12.0 {
+ t.Fatalf("d1 series = %v, want [11 12]", v)
+ }
+ if v := byDevice["d2"]; len(v) != 2 || v[0] != 21.0 || v[1] != 22.5 {
+ t.Fatalf("d2 series = %v, want [21 22.5]", v)
+ }
+}
+
+// TestBuildTableResponseFrameTableFormatPreservesOrder pins that the Table
+// format neither re-sorts rows (the user's ORDER BY wins) nor pivots tags
+// into labels.
+func TestBuildTableResponseFrameTableFormatPreservesOrder(t *testing.T) {
+ ds := &tableQueryDataSet{
+ ColumnNames: []string{"time", "device", "temperature"},
+ DataTypes: []string{"TIMESTAMP", "STRING", "DOUBLE"},
+ Values: [][]interface{}{
+ {ts(2000), "d2", float64(22.5)},
+ {ts(1000), "d1", float64(11.0)},
+ },
+ }
+
+ frame := buildTableResponseFrame(ds, tableFormatTable)
+ if len(frame.Fields) != 3 {
+ t.Fatalf("expected 3 plain fields, got %d", len(frame.Fields))
+ }
+ first, ok := frame.Fields[0].At(0).(*time.Time)
+ if !ok || first == nil || !first.Equal(ts(2000)) {
+ t.Fatalf("row order changed: first time = %#v, want t=2000ms",
frame.Fields[0].At(0))
+ }
+ if _, ok := frame.Fields[1].At(0).(*string); !ok {
+ t.Fatalf("tag column should stay a plain string field in Table
format")
+ }
+}
+
+// TestBuildTableResponseFrameNullTimeFallsBack pins the safety net: when the
+// long-to-wide pivot cannot apply (here: a null timestamp), the plain frame is
+// returned instead of an error or a panic.
+func TestBuildTableResponseFrameNullTimeFallsBack(t *testing.T) {
+ ds := &tableQueryDataSet{
+ ColumnNames: []string{"time", "device", "temperature"},
+ DataTypes: []string{"TIMESTAMP", "STRING", "DOUBLE"},
+ Values: [][]interface{}{
+ {ts(1000), "d1", float64(11.0)},
+ {nil, "d2", float64(21.0)},
+ },
+ }
+
+ frame := buildTableResponseFrame(ds, tableFormatTimeSeries)
+ if len(frame.Fields) != 3 {
+ t.Fatalf("expected plain 3-field fallback frame, got %d
fields", len(frame.Fields))
+ }
+ if frame.TimeSeriesSchema().Type != data.TimeSeriesTypeLong {
+ t.Fatalf("fallback frame should still be the long-shaped
original")
+ }
+}
+
+// TestVerifyQueryTableModel pins the table-mode input validation: SQL and
+// DATABASE are both required (the latter so every query USEs its database on
+// the pooled session and leftover USE state can never leak between panels).
+func TestVerifyQueryTableModel(t *testing.T) {
+ build := func(body string) backend.DataQuery {
+ return backend.DataQuery{JSON: []byte(body)}
+ }
+
+ if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table
Model","sql":"SELECT 1","database":"db1"}`)); msg != "" {
+ t.Fatalf("valid table query rejected: %q", msg)
+ }
+ if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table Model","sql":"
","database":"db1"}`)); msg != "Input error, SQL is required" {
+ t.Fatalf("blank sql not rejected: %q", msg)
+ }
+ if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table
Model","sql":"SELECT 1","database":" "}`)); msg != "Input error, DATABASE is
required" {
+ t.Fatalf("blank database not rejected: %q", msg)
+ }
+ if _, msg := verifyQuery(build(`{"sqlType":"SQL: Table
Model","sql":"SELECT 1"}`)); msg != "Input error, DATABASE is required" {
+ t.Fatalf("missing database not rejected: %q", msg)
+ }
+}
+
+func TestTableRPCEndpoint(t *testing.T) {
+ cases := []struct {
+ name string
+ rpcAddress string
+ url string
+ wantHost string
+ wantPort string
+ }{
+ {name: "derived from http url", url:
"http://192.168.1.10:18080";, wantHost: "192.168.1.10", wantPort: "6667"},
+ {name: "derived from url with trailing slash", url:
"http://iotdb.example.com:18080/";, wantHost: "iotdb.example.com", wantPort:
"6667"},
+ {name: "derived from bare host and rest port", url:
"192.168.1.10:18080", wantHost: "192.168.1.10", wantPort: "6667"},
+ {name: "explicit host and port", rpcAddress: "10.0.0.5:7777",
url: "http://x:18080";, wantHost: "10.0.0.5", wantPort: "7777"},
+ {name: "explicit bare host gets default port", rpcAddress:
"10.0.0.5", url: "http://x:18080";, wantHost: "10.0.0.5", wantPort: "6667"},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ d := &IoTDBDataSource{Ulr: c.url, RPCAddress:
c.rpcAddress}
+ host, port, err := d.tableRPCEndpoint()
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if host != c.wantHost || port != c.wantPort {
+ t.Fatalf("endpoint = %s:%s, want %s:%s", host,
port, c.wantHost, c.wantPort)
+ }
+ })
+ }
+}
diff --git a/connectors/grafana-plugin/src/ConfigEditor.tsx
b/connectors/grafana-plugin/src/ConfigEditor.tsx
index fed474d..9d3b134 100644
--- a/connectors/grafana-plugin/src/ConfigEditor.tsx
+++ b/connectors/grafana-plugin/src/ConfigEditor.tsx
@@ -42,6 +42,15 @@ export class ConfigEditor extends PureComponent<Props,
State> {
onOptionsChange({ ...options, jsonData });
};
+ onRPCAddressChange = (event: ChangeEvent<HTMLInputElement>) => {
+ const { onOptionsChange, options } = this.props;
+ const jsonData = {
+ ...options.jsonData,
+ rpcAddress: event.target.value,
+ };
+ onOptionsChange({ ...options, jsonData });
+ };
+
// Secure field (only sent to the backend)
onPasswordChange = (event: ChangeEvent<HTMLInputElement>) => {
const { onOptionsChange, options } = this.props;
@@ -101,6 +110,18 @@ export class ConfigEditor extends PureComponent<Props,
State> {
onChange={this.onPasswordChange}
/>
</InlineField>
+ <InlineField
+ label="rpc address"
+ labelWidth={12}
+ tooltip="The IoTDB RPC endpoint (host or host:port) used by the SQL:
Table Model mode's native client. Leave empty to use the URL's host with the
default RPC port 6667."
+ >
+ <Input
+ onChange={this.onRPCAddressChange}
+ value={jsonData.rpcAddress || ''}
+ placeholder="iotdb-host:6667 (optional)"
+ width={40}
+ />
+ </InlineField>
</div>
);
diff --git a/connectors/grafana-plugin/src/QueryEditor.tsx
b/connectors/grafana-plugin/src/QueryEditor.tsx
index 56131bc..9704c46 100644
--- a/connectors/grafana-plugin/src/QueryEditor.tsx
+++ b/connectors/grafana-plugin/src/QueryEditor.tsx
@@ -26,7 +26,7 @@ import { FromValue } from './componments/FromValue';
import { WhereValue } from './componments/WhereValue';
import { ControlValue } from './componments/ControlValue';
import { FillValue } from './componments/FillValue';
-import { Segment } from '@grafana/ui';
+import { Input, Segment, TextArea } from '@grafana/ui';
import { toOption } from './functions';
import { GroupByLabel } from './componments/GroupBy';
@@ -46,6 +46,9 @@ interface State {
isDropDownList: boolean;
sqlType: string;
shouldAdd: boolean;
+ database: string;
+ sql: string;
+ format: string;
}
const selectElement = [
@@ -64,7 +67,8 @@ const selectElement = [
const paths = [''];
const expressions = [''];
-const selectType = ['SQL: Full Customized', 'SQL: Drop-down List'];
+const selectType = ['SQL: Full Customized', 'SQL: Drop-down List', 'SQL: Table
Model'];
+const tableFormats = ['Time series', 'Table'];
const commonOption: SelectableValue<string> = { label: '*', value: '*' };
const commonOptionDou: SelectableValue<string> = { label: '**', value: '**' };
type Props = QueryEditorProps<DataSource, IoTDBQuery, IoTDBOptions>;
@@ -87,6 +91,9 @@ export class QueryEditor extends PureComponent<Props, State> {
isDropDownList: false,
sqlType: selectType[0],
shouldAdd: true,
+ database: '',
+ sql: '',
+ format: tableFormats[0],
};
@@ -134,7 +141,26 @@ export class QueryEditor extends PureComponent<Props,
State> {
onChange({ ...query, groupBy: g });
};
-
+ onDatabaseChange = (event: ChangeEvent<HTMLInputElement>) => {
+ const { onChange, query } = this.props;
+ const database = event.target.value;
+ this.setState({ database });
+ onChange({ ...query, database });
+ };
+
+ onSqlChange = (event: ChangeEvent<HTMLTextAreaElement>) => {
+ const { onChange, query } = this.props;
+ const sql = event.target.value;
+ this.setState({ sql });
+ onChange({ ...query, sql });
+ };
+
+ onFormatChange = ({ value: value = tableFormats[0] }:
SelectableValue<string>) => {
+ const { onChange, query } = this.props;
+ this.setState({ format: value });
+ onChange({ ...query, format: value });
+ };
+
onSelectTypeChange = (event: ChangeEvent<HTMLInputElement>) => {
const { onChange, query } = this.props;
onChange({ ...query });
@@ -175,7 +201,13 @@ export class QueryEditor extends PureComponent<Props,
State> {
componentDidMount() {
if (this.props.query.sqlType) {
- this.setState({ isDropDownList: this.props.query.isDropDownList,
sqlType: this.props.query.sqlType });
+ this.setState({
+ isDropDownList: this.props.query.isDropDownList,
+ sqlType: this.props.query.sqlType,
+ database: this.props.query.database ?? '',
+ sql: this.props.query.sql ?? '',
+ format: this.props.query.format ?? tableFormats[0],
+ });
} else {
this.props.query.sqlType = selectType[0];
}
@@ -231,7 +263,7 @@ export class QueryEditor extends PureComponent<Props,
State> {
condition: '',
});
onChange({ ...query, sqlType: value, isDropDownList: false
});
- } else {
+ } else if (value === selectType[1]) {
this.props.query.sqlType = selectType[1];
this.props.query.expression = [''];
this.props.query.prefixPath = [''];
@@ -247,6 +279,22 @@ export class QueryEditor extends PureComponent<Props,
State> {
control: '',
});
onChange({ ...query, sqlType: value, isDropDownList: true
});
+ } else {
+ this.props.query.sqlType = selectType[2];
+ this.props.query.expression = [''];
+ this.props.query.prefixPath = [''];
+ this.props.query.condition = '';
+ this.props.query.control = '';
+ this.props.query.isDropDownList = false;
+ this.setState({
+ isDropDownList: false,
+ sqlType: selectType[2],
+ expression: [''],
+ prefixPath: [''],
+ condition: '',
+ control: '',
+ });
+ onChange({ ...query, sqlType: value, isDropDownList: false
});
}
}}
options={selectType.map(toOption)}
@@ -254,7 +302,7 @@ export class QueryEditor extends PureComponent<Props,
State> {
className="query-keyword width-10"
/>
</div>
- {!this.state.isDropDownList && (
+ {this.state.sqlType === selectType[0] && (
<>
<div className="gf-form">
<QueryInlineField label={'SELECT'}>
@@ -335,6 +383,39 @@ export class QueryEditor extends PureComponent<Props,
State> {
</div>
</>
)}
+ {this.state.sqlType === selectType[2] && (
+ <>
+ <div className="gf-form">
+ <QueryInlineField label={'DATABASE'}>
+ <Input
+ value={query.database ?? this.state.database}
+ placeholder={'database name (required)'}
+ onChange={this.onDatabaseChange}
+ />
+ </QueryInlineField>
+ </div>
+ <div className="gf-form">
+ <QueryInlineField label={'SQL'}>
+ <TextArea
+ rows={4}
+ value={query.sql ?? this.state.sql}
+ placeholder={'SELECT time, s0 FROM my_table WHERE
$__timeFilter(time)'}
+ onChange={this.onSqlChange}
+ />
+ </QueryInlineField>
+ </div>
+ <div className="gf-form">
+ <QueryInlineField label={'FORMAT'}>
+ <Segment
+ onChange={this.onFormatChange}
+ options={tableFormats.map(toOption)}
+ value={query.format ?? this.state.format}
+ className="query-keyword width-10"
+ />
+ </QueryInlineField>
+ </div>
+ </>
+ )}
</>
}
</>
diff --git a/connectors/grafana-plugin/src/datasource.ts
b/connectors/grafana-plugin/src/datasource.ts
index 4fff0e8..319f804 100644
--- a/connectors/grafana-plugin/src/datasource.ts
+++ b/connectors/grafana-plugin/src/datasource.ts
@@ -126,6 +126,13 @@ export class DataSource extends
DataSourceWithBackend<IoTDBQuery, IoTDBOptions>
if (query.fillClauses) {
query.fillClauses = getTemplateSrv().replace(query.fillClauses,
scopedVars);
}
+ } else if (query.sqlType === 'SQL: Table Model') {
+ if (query.sql) {
+ query.sql = getTemplateSrv().replace(query.sql, scopedVars);
+ }
+ if (query.database) {
+ query.database = getTemplateSrv().replace(query.database, scopedVars);
+ }
}
return query;
}
diff --git a/connectors/grafana-plugin/src/types.ts
b/connectors/grafana-plugin/src/types.ts
index a67aa42..8f793f5 100644
--- a/connectors/grafana-plugin/src/types.ts
+++ b/connectors/grafana-plugin/src/types.ts
@@ -33,6 +33,13 @@ export interface IoTDBQuery extends DataQuery {
limitAll?: LimitAll;
options: Array<Array<SelectableValue<string>>>;
hide: boolean;
+
+ // Table-model mode: a standard SQL statement run against `database`,
+ // rendered either as time series (long results are pivoted into one series
+ // per tag combination) or as a plain table.
+ database?: string;
+ sql?: string;
+ format?: string;
}
export interface GroupBy {
@@ -58,6 +65,10 @@ export interface LimitAll {
export interface IoTDBOptions extends DataSourceJsonData {
url: string;
username: string;
+ // The IoTDB RPC endpoint (host or host:port) used by the table-model mode's
+ // native client. When empty, the URL's host with the default RPC port 6667
+ // is used.
+ rpcAddress?: string;
}
/**
![https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/add%20empty%20panel.png?raw=true"](https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/add%20empty%20panel.png?raw=true"){kind=link}
![https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/grafana_input_style.png?raw=true"](https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/grafana_input_style.png?raw=true"){kind=link}
![https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/grafana_input2.png?raw=true"](https://github.com/apache/iotdb-bin-resources/blob/main/docs/UserGuide/Ecosystem%20Integration/Grafana-plugin/grafana_input2.png?raw=true"){kind=link}