This is an automated email from the ASF dual-hosted git repository.
JackieTien97 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 28f3032 feat: add support for query type options in IoTDB queries
(#129)
28f3032 is described below
commit 28f3032fee1e3b8a6c3f41400ca7bb28e68307aa
Author: Zhao Xinqi <[email protected]>
AuthorDate: Fri Aug 28 09:14:38 2026 +0800
feat: add support for query type options in IoTDB queries (#129)
---
connectors/grafana-plugin/README.md | 7 +
connectors/grafana-plugin/pkg/plugin/plugin.go | 2 +
.../grafana-plugin/pkg/plugin/table_query.go | 149 +++++++++-
.../grafana-plugin/pkg/plugin/table_query_test.go | 329 +++++++++++++++++++++
connectors/grafana-plugin/src/QueryEditor.tsx | 18 +-
connectors/grafana-plugin/src/datasource.test.ts | 26 +-
connectors/grafana-plugin/src/types.ts | 25 ++
7 files changed, 552 insertions(+), 4 deletions(-)
diff --git a/connectors/grafana-plugin/README.md
b/connectors/grafana-plugin/README.md
index ccd5948..c6e31ec 100644
--- a/connectors/grafana-plugin/README.md
+++ b/connectors/grafana-plugin/README.md
@@ -155,9 +155,16 @@ SQL input box: a single table-model SELECT statement. The
following macros are e
| `$__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 |
+| `$__evaluationTime`, `$__evaluationTime()` | the panel range end as an ISO
8601 UTC timestamp literal; use this explicitly for point-in-time constructs
such as `HOP(... ORIGIN => ...)` |
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.
+QUERY TYPE option:
+
+- `Range` (default): evaluates the SQL over the complete Grafana time range.
Queries saved before this option existed remain Range queries.
+- `Instant`: evaluates the SQL over the complete Grafana range, then returns
the latest row at or before the range end for each series, with its timestamp
normalized to that evaluation time. This supports samples whose timestamps do
not exactly equal the panel end. SQL that needs a window boundary aligned to
the evaluation time (for example `HOP(... ORIGIN => ...)`) must explicitly use
`$__evaluationTime`; the plugin does not rewrite `ORIGIN` values, comments, or
string literals.
+- `Both`: executes Range and Instant independently and returns both results in
distinctly named frames.
+
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.
diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go
b/connectors/grafana-plugin/pkg/plugin/plugin.go
index e31e048..970b7a6 100644
--- a/connectors/grafana-plugin/pkg/plugin/plugin.go
+++ b/connectors/grafana-plugin/pkg/plugin/plugin.go
@@ -183,6 +183,8 @@ type queryParam struct {
Format string `json:"format"`
IntervalMS int64 `json:"-"`
LegendFormat string `json:"legendFormat"`
+ Instant bool `json:"instant"`
+ Range bool `json:"range"`
}
type QueryDataReq struct {
diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go
b/connectors/grafana-plugin/pkg/plugin/table_query.go
index b03e3cc..b42ede1 100644
--- a/connectors/grafana-plugin/pkg/plugin/table_query.go
+++ b/connectors/grafana-plugin/pkg/plugin/table_query.go
@@ -73,6 +73,11 @@ var timeFilterRe =
regexp.MustCompile(`\$__timeFilter\(\s*((?:[^()]|\([^()]*\))*
var (
timeFromRe = regexp.MustCompile(`\$__timeFrom\b(?:\s*\(\s*\))?`)
timeToRe = regexp.MustCompile(`\$__timeTo\b(?:\s*\(\s*\))?`)
+ // $__evaluationTime is an explicit opt-in for SQL constructs such as
HOP
+ // ORIGIN that need Grafana's point-in-time evaluation bound. It is
deliberately
+ // separate from $__timeFrom/__timeTo so Instant does not alter
ordinary range
+ // predicates or user-authored constants.
+ evaluationTimeRe =
regexp.MustCompile(`\$__evaluationTime\b(?:\s*\(\s*\))?`)
// These patterns intentionally match the macro prefix.
hasStandaloneMacro
// and replaceStandaloneMacro reject an identifier byte after the
match, so
// $__interval_ms is not mistaken for $__interval.
@@ -129,6 +134,7 @@ func (d *IoTDBDataSource) expandVariableMacros(sql string)
string {
// $__timeFilter(col) -> (col >= <from> AND col <= <to>)
// $__timeFrom[()] -> <from>
// $__timeTo[()] -> <to>
+// $__evaluationTime[()] -> <to>
// $__interval -> a fixed-width IoTDB duration literal
// $__interval_ms -> the interval in milliseconds, per Grafana's
contract
//
@@ -156,6 +162,7 @@ func expandTableMacros(sql string, startMs int64, endMs
int64, intervalMS int64)
from := formatTimeLiteral(startMs)
to := formatTimeLiteral(endMs)
+ sql = replaceStandaloneMacro(sql, evaluationTimeRe, to)
sql = timeFilterRe.ReplaceAllStringFunc(sql, func(m string) string {
col := strings.TrimSpace(timeFilterRe.FindStringSubmatch(m)[1])
if col == "" {
@@ -168,6 +175,14 @@ func expandTableMacros(sql string, startMs int64, endMs
int64, intervalMS int64)
return sql, nil
}
+// expandInstantTableMacros expands an Instant query over the complete Grafana
+// range, while exposing the range end through the explicit $__evaluationTime
+// macro. The result selector later chooses the latest row at or before that
+// evaluation time for each series.
+func expandInstantTableMacros(sql string, startMs int64, endMs int64,
intervalMS int64) (string, error) {
+ return expandTableMacros(sql, startMs, endMs, intervalMS)
+}
+
// hasStandaloneMacro reports whether re has a match that is not followed by an
// identifier character. Go's regexp package intentionally has no lookahead,
// so the boundary check is performed while scanning matches.
@@ -333,6 +348,30 @@ func (d *IoTDBDataSource) getTablePool()
(*client.TableSessionPool, error) {
// 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 {
+ wantsInstant := qp.Instant
+ wantsRange := !qp.Instant || qp.Range
+ if wantsInstant && wantsRange {
+ response := backend.DataResponse{}
+ rangeQuery := *qp
+ rangeQuery.Instant = false
+ rangeResponse := d.queryTableModelOnce(ctx, &rangeQuery,
"response_range")
+ response.Frames = append(response.Frames,
rangeResponse.Frames...)
+ if rangeResponse.Error != nil {
+ response.Error = rangeResponse.Error
+ return response
+ }
+
+ instantQuery := *qp
+ instantResponse := d.queryTableModelOnce(ctx, &instantQuery,
"response_instant")
+ response.Frames = append(response.Frames,
instantResponse.Frames...)
+ response.Error = instantResponse.Error
+ return response
+ }
+
+ return d.queryTableModelOnce(ctx, qp, "response")
+}
+
+func (d *IoTDBDataSource) queryTableModelOnce(ctx context.Context, qp
*queryParam, frameName string) backend.DataResponse {
response := backend.DataResponse{}
if (hasStandaloneMacro(qp.Sql, intervalRe) ||
hasStandaloneMacro(qp.Sql, intervalMSRe)) && qp.IntervalMS <= 0 {
@@ -351,6 +390,9 @@ func (d *IoTDBDataSource) queryTableModel(ctx
context.Context, qp *queryParam) b
response.Error = err
return response
}
+ if qp.Instant {
+ dataSet = prepareInstantDataSet(dataSet, qp.EndTime)
+ }
if !strings.EqualFold(qp.Format, tableFormatTable) &&
!hasPlottableValue(dataSet) {
// Time Series with no plottable values — zero rows, or rows
whose value
@@ -360,20 +402,123 @@ func (d *IoTDBDataSource) queryTableModel(ctx
context.Context, qp *queryParam) b
return response
}
- response.Frames = append(response.Frames,
buildTableResponseFrame(dataSet, qp.Format, qp.LegendFormat))
+ frame := buildTableResponseFrame(dataSet, qp.Format, qp.LegendFormat)
+ frame.Name = frameName
+ response.Frames = append(response.Frames, frame)
return response
}
// executeTableQuery runs and fetches one table-model query. queryTableModel
// owns response semantics so the same zero-row path is covered in tests.
func (d *IoTDBDataSource) executeTableQuery(ctx context.Context, qp
*queryParam) (*tableQueryDataSet, error) {
- sql, err := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime,
qp.IntervalMS)
+ var sql string
+ var err error
+ if qp.Instant {
+ sql, err = expandInstantTableMacros(qp.Sql, qp.StartTime,
qp.EndTime, qp.IntervalMS)
+ } else {
+ sql, err = expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime,
qp.IntervalMS)
+ }
if err != nil {
return nil, err
}
return d.executeTableStatement(ctx, qp.Database, sql)
}
+// prepareInstantDataSet enforces the point-in-time response contract after the
+// SQL has evaluated over the requested range. For each label/series identity
it
+// keeps the latest row whose timestamp is at or before evaluationMs, then
+// normalizes that row's timestamp to evaluationMs. Scalar results without a
+// TIMESTAMP column receive a synthetic evaluation-time column.
+func prepareInstantDataSet(dataSet *tableQueryDataSet, evaluationMs int64)
*tableQueryDataSet {
+ evaluation := time.UnixMilli(evaluationMs).UTC()
+ timeCol := -1
+ for i, dataType := range dataSet.DataTypes {
+ if strings.EqualFold(dataType, "TIMESTAMP") {
+ timeCol = i
+ break
+ }
+ }
+
+ prepared := &tableQueryDataSet{
+ ColumnNames: append([]string(nil), dataSet.ColumnNames...),
+ DataTypes: append([]string(nil), dataSet.DataTypes...),
+ Values: make([][]interface{}, 0, len(dataSet.Values)),
+ }
+ if timeCol < 0 {
+ seenSeries := make(map[string]struct{}, len(dataSet.Values))
+ prepared.ColumnNames = append([]string{"time"},
prepared.ColumnNames...)
+ prepared.DataTypes = append([]string{"TIMESTAMP"},
prepared.DataTypes...)
+ for _, row := range dataSet.Values {
+ seriesKey := instantSeriesKey(dataSet, row, timeCol)
+ if _, exists := seenSeries[seriesKey]; exists {
+ continue
+ }
+ seenSeries[seriesKey] = struct{}{}
+ instantRow := make([]interface{}, 0, len(row)+1)
+ instantRow = append(instantRow, evaluation)
+ instantRow = append(instantRow, row...)
+ prepared.Values = append(prepared.Values, instantRow)
+ }
+ return prepared
+ }
+
+ // Keep the first-seen series order while replacing each series'
candidate
+ // whenever a later eligible timestamp is encountered. A row after the
+ // evaluation time can never satisfy Instant semantics.
+ type candidate struct {
+ row []interface{}
+ time time.Time
+ }
+ candidates := make(map[string]candidate, len(dataSet.Values))
+ seriesOrder := make([]string, 0, len(dataSet.Values))
+ for _, row := range dataSet.Values {
+ rowTime, ok := rowTimeAt(row, timeCol)
+ if !ok || rowTime.After(evaluation) {
+ continue
+ }
+ seriesKey := instantSeriesKey(dataSet, row, timeCol)
+ current, exists := candidates[seriesKey]
+ if !exists {
+ seriesOrder = append(seriesOrder, seriesKey)
+ } else if !rowTime.After(current.time) {
+ // Preserve the first row when timestamps tie.
+ continue
+ }
+ candidates[seriesKey] = candidate{row: row, time: rowTime}
+ }
+ for _, seriesKey := range seriesOrder {
+ row := candidates[seriesKey].row
+ instantRow := append([]interface{}(nil), row...)
+ instantRow[timeCol] = evaluation
+ prepared.Values = append(prepared.Values, instantRow)
+ }
+ return prepared
+}
+
+// instantSeriesKey mirrors the long-to-wide series identity: non-time,
+// non-numeric columns are labels, while each numeric column is a value series.
+// Grouping rows by this key lets Instant select one latest eligible point for
+// each series rather than treating the query's globally last row as its
result.
+func instantSeriesKey(dataSet *tableQueryDataSet, row []interface{}, timeCol
int) string {
+ var key strings.Builder
+ for col, cell := range row {
+ if col == timeCol || col < len(dataSet.DataTypes) &&
isNumericTableType(dataSet.DataTypes[col]) {
+ continue
+ }
+ value := "<nil>"
+ if cell != nil {
+ value = fmt.Sprintf("%T:%s", cell, toString(cell))
+ }
+ key.WriteString(strconv.Itoa(col))
+ key.WriteByte(':')
+ key.WriteString(strconv.Itoa(len(value)))
+ key.WriteByte(':')
+ key.WriteString(value)
+ key.WriteByte(';')
+ }
+ return key.String()
+}
+
// executeTableStatement runs a table-model SQL statement against the given
// database on a pooled native-client session and returns the fetched dataset.
// The database is USEd on the session so a statement's table references
resolve
diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go
b/connectors/grafana-plugin/pkg/plugin/table_query_test.go
index ad36944..44d97ee 100644
--- a/connectors/grafana-plugin/pkg/plugin/table_query_test.go
+++ b/connectors/grafana-plugin/pkg/plugin/table_query_test.go
@@ -33,6 +33,18 @@ func ts(ms int64) time.Time {
return time.UnixMilli(ms).UTC()
}
+func fieldTime(value interface{}) (time.Time, bool) {
+ switch value := value.(type) {
+ case time.Time:
+ return value, true
+ case *time.Time:
+ if value != nil {
+ return *value, true
+ }
+ }
+ return time.Time{}, false
+}
+
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
@@ -64,6 +76,11 @@ func TestExpandTableMacros(t *testing.T) {
in: "SELECT * FROM db.t WHERE time >= $__timeFrom()
AND time <= $__timeTo( )",
want: "SELECT * FROM db.t WHERE time >= " + fromLit + "
AND time <= " + toLit,
},
+ {
+ name: "evaluation time uses range end in either form",
+ in: "SELECT $__evaluationTime, $__evaluationTime()
FROM db.t",
+ want: "SELECT " + toLit + ", " + toLit + " FROM db.t",
+ },
{
name: "timeFilter with one nested paren level",
in: "SELECT * FROM db.t WHERE $__timeFilter(cast(x))",
@@ -930,3 +947,315 @@ func TestQueryParamLegendFormatEmpty(t *testing.T) {
t.Fatalf("LegendFormat should default to empty, got %q",
qp.LegendFormat)
}
}
+
+func TestQueryParamInstantDeserialization(t *testing.T) {
+ qp, msg := verifyQuery(backend.DataQuery{JSON: []byte(`{"sqlType":"SQL:
Table Model","sql":"SELECT 1","database":"db1","instant":true}`)})
+ if msg != "" {
+ t.Fatalf("valid query rejected: %q", msg)
+ }
+ if !qp.Instant || qp.Range {
+ t.Fatalf("instant flags = instant:%v range:%v, want true and
false", qp.Instant, qp.Range)
+ }
+}
+
+func TestQueryParamOmittedInstantIsBackwardCompatibleRange(t *testing.T) {
+ qp, msg := verifyQuery(backend.DataQuery{JSON: []byte(`{"sqlType":"SQL:
Table Model","sql":"SELECT 1","database":"db1"}`)})
+ if msg != "" {
+ t.Fatalf("valid query rejected: %q", msg)
+ }
+ if qp.Instant || qp.Range {
+ t.Fatalf("omitted query flags = instant:%v range:%v, want false
and false", qp.Instant, qp.Range)
+ }
+}
+
+func
TestExpandInstantTableMacrosPreservesRangeAndExpandsExplicitEvaluationTime(t
*testing.T) {
+ const start int64 = 1600000000000
+ const evaluation int64 = 1600000001000
+ const startLit = "2020-09-13T12:26:40.000+00:00"
+ const evaluationLit = "2020-09-13T12:26:41.000+00:00"
+ sql := `SELECT window_start AS time
+FROM HOP(
+ DATA => (SELECT time FROM metrics WHERE time >= $__timeFrom - 300000 AND
time <= $__timeTo),
+ TIMECOL => 'time', SLIDE => $__interval, SIZE => 5m, ORIGIN =>
$__evaluationTime)
+WHERE $__timeFilter(window_start)
+-- ORIGIN => 0 must remain a comment
+/* 'ORIGIN => 0' must remain a string */`
+
+ got, err := expandInstantTableMacros(sql, start, evaluation, 60000)
+ if err != nil {
+ t.Fatalf("expandInstantTableMacros() unexpected error: %v", err)
+ }
+ for _, want := range []string{
+ "time >= " + startLit + " - 300000",
+ "time <= " + evaluationLit,
+ "SLIDE => 1m",
+ "ORIGIN => " + evaluationLit,
+ "window_start >= " + startLit + " AND window_start <= " +
evaluationLit,
+ "-- ORIGIN => 0 must remain a comment",
+ "/* 'ORIGIN => 0' must remain a string */",
+ } {
+ if !strings.Contains(got, want) {
+ t.Fatalf("instant SQL does not contain %q:\n%s", want,
got)
+ }
+ }
+ if strings.Contains(got, "$__") {
+ t.Fatalf("instant SQL contains an unexpanded macro: %s", got)
+ }
+}
+
+func TestExpandInstantTableMacrosDoesNotRewriteUserOrigin(t *testing.T) {
+ const start int64 = 1600000000000
+ const evaluation int64 = 1600000001000
+ sql := `SELECT window_start AS time
+FROM HOP(DATA => metrics, SLIDE => $__interval, SIZE => 5m, ORIGIN => 0)
+WHERE $__timeFilter(window_start)
+-- ORIGIN => 0
+/* literal: 'ORIGIN => 0' */`
+
+ got, err := expandInstantTableMacros(sql, start, evaluation, 60000)
+ if err != nil {
+ t.Fatalf("expandInstantTableMacros() unexpected error: %v", err)
+ }
+ if count := strings.Count(got, "ORIGIN => 0"); count != 3 {
+ t.Fatalf("ORIGIN => 0 occurrences = %d, want 3 unchanged
occurrences:\n%s", count, got)
+ }
+}
+
+func TestExecuteTableQueryInstantKeepsOrdinaryTableSQLRange(t *testing.T) {
+ var executedSQL string
+ d := &IoTDBDataSource{
+ tableExecutor: func(_ context.Context, _ string, sql string)
(*tableQueryDataSet, error) {
+ executedSQL = sql
+ return &tableQueryDataSet{}, nil
+ },
+ }
+ _, err := d.executeTableQuery(context.Background(), &queryParam{
+ Sql: "SELECT time, value FROM metrics WHERE
$__timeFilter(time)",
+ Database: "db1",
+ StartTime: 1600000000000,
+ EndTime: 1600000001000,
+ Instant: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := "SELECT time, value FROM metrics WHERE (time >=
2020-09-13T12:26:40.000+00:00 AND time <= 2020-09-13T12:26:41.000+00:00)"
+ if executedSQL != want {
+ t.Fatalf("ordinary Instant SQL = %q, want complete range %q",
executedSQL, want)
+ }
+}
+
+func TestExecuteTableQueryInstantSendsRangeSQLAndExplicitEvaluationTime(t
*testing.T) {
+ var executedSQL string
+ d := &IoTDBDataSource{
+ tableExecutor: func(_ context.Context, _ string, sql string)
(*tableQueryDataSet, error) {
+ executedSQL = sql
+ return &tableQueryDataSet{}, nil
+ },
+ }
+ _, err := d.executeTableQuery(context.Background(), &queryParam{
+ Sql: "SELECT window_start AS time FROM HOP(DATA => t,
SLIDE => $__interval, SIZE => 5m, ORIGIN => $__evaluationTime) WHERE
$__timeFilter(window_start)",
+ Database: "db1",
+ StartTime: 1600000000000,
+ EndTime: 1600000001000,
+ IntervalMS: 1000,
+ Instant: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ const startLit = "2020-09-13T12:26:40.000+00:00"
+ const evaluationLit = "2020-09-13T12:26:41.000+00:00"
+ if !strings.Contains(executedSQL, "window_start >= "+startLit+" AND
window_start <= "+evaluationLit) {
+ t.Fatalf("instant SQL did not retain the complete time range:
%s", executedSQL)
+ }
+ if !strings.Contains(executedSQL, "ORIGIN => "+evaluationLit) {
+ t.Fatalf("instant SQL did not expand the explicit evaluation
macro: %s", executedSQL)
+ }
+}
+
+func TestQueryTableModelRangeStillReturnsMultiplePoints(t *testing.T) {
+ d := &IoTDBDataSource{
+ tableQueryRunner: func(context.Context, *queryParam)
(*tableQueryDataSet, error) {
+ return &tableQueryDataSet{
+ ColumnNames: []string{"time", "value"},
+ DataTypes: []string{"TIMESTAMP", "DOUBLE"},
+ Values: [][]interface{}{
+ {ts(1000), float64(1)},
+ {ts(2000), float64(2)},
+ },
+ }, nil
+ },
+ }
+
+ response := d.queryTableModel(context.Background(), &queryParam{Format:
tableFormatTimeSeries, EndTime: 2000})
+ if response.Error != nil || len(response.Frames) != 1 {
+ t.Fatalf("range response = frames:%d error:%v",
len(response.Frames), response.Error)
+ }
+ if got := response.Frames[0].Fields[0].Len(); got != 2 {
+ t.Fatalf("range time points = %d, want 2", got)
+ }
+}
+
+func TestQueryTableModelInstantReturnsOnePointPerSeriesAtQueryEnd(t
*testing.T) {
+ const evaluation int64 = 3000
+ d := &IoTDBDataSource{
+ tableQueryRunner: func(context.Context, *queryParam)
(*tableQueryDataSet, error) {
+ return &tableQueryDataSet{
+ ColumnNames: []string{"time", "instance",
"value"},
+ DataTypes: []string{"TIMESTAMP", "STRING",
"DOUBLE"},
+ Values: [][]interface{}{
+ {ts(1000), "node-a", float64(9)},
+ {ts(evaluation), "node-a", float64(0)},
+ {ts(evaluation), "node-a", float64(99)},
+ {ts(2000), "node-b", float64(8)},
+ {ts(evaluation), "node-b", float64(2)},
+ },
+ }, nil
+ },
+ }
+
+ response := d.queryTableModel(context.Background(), &queryParam{
+ Format: tableFormatTimeSeries,
+ LegendFormat: "{{instance}}",
+ Instant: true,
+ EndTime: evaluation,
+ })
+ if response.Error != nil || len(response.Frames) != 1 {
+ t.Fatalf("instant response = frames:%d error:%v",
len(response.Frames), response.Error)
+ }
+ frame := response.Frames[0]
+ if len(frame.Fields) != 3 {
+ t.Fatalf("instant fields = %d, want time plus two labeled
series", len(frame.Fields))
+ }
+ if got := frame.Fields[0].Len(); got != 1 {
+ t.Fatalf("instant timestamps = %d, want 1", got)
+ }
+ gotTime, ok := fieldTime(frame.Fields[0].At(0))
+ if !ok || !gotTime.Equal(ts(evaluation)) {
+ t.Fatalf("instant timestamp = %#v, want %v",
frame.Fields[0].At(0), ts(evaluation))
+ }
+
+ values := map[string]float64{}
+ for _, field := range frame.Fields[1:] {
+ if field.Len() != 1 {
+ t.Fatalf("series %q has %d points, want 1", field.Name,
field.Len())
+ }
+ value, ok := field.At(0).(*float64)
+ if !ok || value == nil {
+ t.Fatalf("series %q value = %#v, want non-null float",
field.Name, field.At(0))
+ }
+ values[field.Labels["instance"]] = *value
+ if field.Config == nil || field.Config.DisplayNameFromDS !=
field.Labels["instance"] {
+ t.Fatalf("series labels/legend not preserved: labels=%v
config=%#v", field.Labels, field.Config)
+ }
+ }
+ if values["node-a"] != 0 || values["node-b"] != 2 {
+ t.Fatalf("instant values = %v, want node-a=0 and node-b=2",
values)
+ }
+}
+
+func TestQueryTableModelInstantSelectsLatestSampleAtOrBeforeQueryEnd(t
*testing.T) {
+ d := &IoTDBDataSource{
+ tableQueryRunner: func(context.Context, *queryParam)
(*tableQueryDataSet, error) {
+ return &tableQueryDataSet{
+ ColumnNames: []string{"time", "value"},
+ DataTypes: []string{"TIMESTAMP", "DOUBLE"},
+ Values: [][]interface{}{
+ {ts(2999), float64(7)},
+ {ts(3001), float64(9)},
+ },
+ }, nil
+ },
+ }
+ response := d.queryTableModel(context.Background(), &queryParam{Format:
tableFormatTimeSeries, Instant: true, EndTime: 3000})
+ if response.Error != nil || len(response.Frames) != 1 {
+ t.Fatalf("Instant response = frames:%d error:%v",
len(response.Frames), response.Error)
+ }
+ frame := response.Frames[0]
+ gotTime, timeOK := fieldTime(frame.Fields[0].At(0))
+ value, valueOK := frame.Fields[1].At(0).(*float64)
+ if !timeOK || !gotTime.Equal(ts(3000)) || !valueOK || value == nil ||
*value != 7 {
+ t.Fatalf("Instant point = time:%v value:%v, want time:%v
value:7", gotTime, value, ts(3000))
+ }
+}
+
+func TestQueryTableModelInstantEmptyAndNullAreNoData(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ values [][]interface{}
+ }{
+ {name: "empty", values: [][]interface{}{}},
+ {name: "null", values: [][]interface{}{{ts(3000), nil}}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ d := &IoTDBDataSource{
+ tableQueryRunner: func(context.Context,
*queryParam) (*tableQueryDataSet, error) {
+ return &tableQueryDataSet{
+ ColumnNames: []string{"time",
"value"},
+ DataTypes:
[]string{"TIMESTAMP", "DOUBLE"},
+ Values: tc.values,
+ }, nil
+ },
+ }
+ response := d.queryTableModel(context.Background(),
&queryParam{Format: tableFormatTimeSeries, Instant: true, EndTime: 3000})
+ if len(response.Frames) != 0 || response.Error != nil {
+ t.Fatalf("%s Instant response = frames:%d
error:%v, want No data", tc.name, len(response.Frames), response.Error)
+ }
+ })
+ }
+}
+
+func TestQueryTableModelInstantAddsEvaluationTimeToScalarResult(t *testing.T) {
+ d := &IoTDBDataSource{
+ tableQueryRunner: func(context.Context, *queryParam)
(*tableQueryDataSet, error) {
+ return &tableQueryDataSet{
+ ColumnNames: []string{"value"},
+ DataTypes: []string{"INT64"},
+ Values: [][]interface{}{{int64(0)}},
+ }, nil
+ },
+ }
+ response := d.queryTableModel(context.Background(), &queryParam{Format:
tableFormatTimeSeries, Instant: true, EndTime: 3000})
+ if len(response.Frames) != 1 || len(response.Frames[0].Fields) != 2 {
+ t.Fatalf("scalar Instant response = %#v", response)
+ }
+ timeValue, timeOK := fieldTime(response.Frames[0].Fields[0].At(0))
+ intValue := response.Frames[0].Fields[1].At(0).(*int64)
+ if !timeOK || !timeValue.Equal(ts(3000)) || intValue == nil ||
*intValue != 0 {
+ t.Fatalf("scalar Instant values = time:%v value:%v", timeValue,
intValue)
+ }
+}
+
+func TestQueryTableModelBothReturnsDistinctRangeAndInstantFrames(t *testing.T)
{
+ var modes []bool
+ d := &IoTDBDataSource{
+ tableQueryRunner: func(_ context.Context, qp *queryParam)
(*tableQueryDataSet, error) {
+ modes = append(modes, qp.Instant)
+ values := [][]interface{}{{ts(1000), float64(1)},
{ts(3000), float64(3)}}
+ return &tableQueryDataSet{
+ ColumnNames: []string{"time", "value"},
+ DataTypes: []string{"TIMESTAMP", "DOUBLE"},
+ Values: values,
+ }, nil
+ },
+ }
+ response := d.queryTableModel(context.Background(), &queryParam{
+ Format: tableFormatTimeSeries,
+ Instant: true,
+ Range: true,
+ EndTime: 3000,
+ })
+ if response.Error != nil || len(response.Frames) != 2 {
+ t.Fatalf("Both response = frames:%d error:%v",
len(response.Frames), response.Error)
+ }
+ if len(modes) != 2 || modes[0] || !modes[1] {
+ t.Fatalf("Both execution modes = %v, want [range instant]",
modes)
+ }
+ if response.Frames[0].Name != "response_range" ||
response.Frames[1].Name != "response_instant" {
+ t.Fatalf("Both frame names = %q, %q", response.Frames[0].Name,
response.Frames[1].Name)
+ }
+ if response.Frames[0].Fields[0].Len() != 2 ||
response.Frames[1].Fields[0].Len() != 1 {
+ t.Fatalf("Both point counts = range:%d instant:%d",
response.Frames[0].Fields[0].Len(), response.Frames[1].Fields[0].Len())
+ }
+}
diff --git a/connectors/grafana-plugin/src/QueryEditor.tsx
b/connectors/grafana-plugin/src/QueryEditor.tsx
index 7f3684c..70ded1a 100644
--- a/connectors/grafana-plugin/src/QueryEditor.tsx
+++ b/connectors/grafana-plugin/src/QueryEditor.tsx
@@ -18,7 +18,7 @@ import defaults from 'lodash/defaults';
import React, { ChangeEvent, PureComponent } from 'react';
import { QueryEditorProps, SelectableValue } from '@grafana/data';
import { DataSource } from './datasource';
-import { GroupBy, IoTDBOptions, IoTDBQuery } from './types';
+import { applyQueryType, getQueryType, GroupBy, IoTDBOptions, IoTDBQuery,
QueryType, queryTypes } from './types';
import { QueryField, QueryInlineField } from './componments/Form';
import { TimeSeries } from './componments/TimeSeries';
import { SelectValue } from './componments/SelectValue';
@@ -70,6 +70,7 @@ const paths = [''];
const expressions = [''];
const selectType = ['SQL: Full Customized', 'SQL: Drop-down List', 'SQL: Table
Model'];
const tableFormats = ['Time series', 'Table'];
+const queryTypeOptions: Array<SelectableValue<QueryType>> =
queryTypes.map((value) => ({ label: value, value }));
const commonOption: SelectableValue<string> = { label: '*', value: '*' };
const commonOptionDou: SelectableValue<string> = { label: '**', value: '**' };
type Props = QueryEditorProps<DataSource, IoTDBQuery, IoTDBOptions>;
@@ -163,6 +164,11 @@ export class QueryEditor extends PureComponent<Props,
State> {
onChange({ ...query, format: value });
};
+ onQueryTypeChange = ({ value: value = queryTypes[0] }:
SelectableValue<QueryType>) => {
+ const { onChange, query } = this.props;
+ onChange(applyQueryType(query, value));
+ };
+
onLegendFormatChange = (event: ChangeEvent<HTMLInputElement>) => {
const { onChange, query } = this.props;
const legendFormat = event.target.value;
@@ -395,6 +401,16 @@ export class QueryEditor extends PureComponent<Props,
State> {
)}
{this.state.sqlType === selectType[2] && (
<>
+ <div className="gf-form">
+ <QueryInlineField label={'QUERY TYPE'}>
+ <Segment
+ onChange={this.onQueryTypeChange}
+ options={queryTypeOptions}
+ value={getQueryType(query)}
+ className="query-keyword width-10"
+ />
+ </QueryInlineField>
+ </div>
<div className="gf-form">
<QueryInlineField label={'DATABASE'}>
<Input
diff --git a/connectors/grafana-plugin/src/datasource.test.ts
b/connectors/grafana-plugin/src/datasource.test.ts
index 6d32cfa..4012eb3 100644
--- a/connectors/grafana-plugin/src/datasource.test.ts
+++ b/connectors/grafana-plugin/src/datasource.test.ts
@@ -15,7 +15,7 @@
* limitations under the License.
*/
import { DataSource } from './datasource';
-import { IoTDBQuery } from './types';
+import { applyQueryType, getQueryType, IoTDBQuery } from './types';
import { ScopedVars } from '@grafana/data';
const mockReplace = jest.fn();
@@ -277,3 +277,27 @@ describe('DataSource', () => {
});
});
});
+
+describe('query type model', () => {
+ const query = { sqlType: 'SQL: Table Model' } as IoTDBQuery;
+
+ it('keeps omitted query flags backward-compatible as Range', () => {
+ expect(getQueryType(query)).toBe('Range');
+ });
+
+ it.each([
+ ['Range', false, true],
+ ['Instant', true, false],
+ ['Both', true, true],
+ ] as const)('serializes %s into dashboard query flags', (queryType, instant,
range) => {
+ const result = applyQueryType(query, queryType);
+
+ expect(result.instant).toBe(instant);
+ expect(result.range).toBe(range);
+ expect(getQueryType(result)).toBe(queryType);
+ });
+
+ it('recognizes an existing dashboard instant target without a range field',
() => {
+ expect(getQueryType({ ...query, instant: true })).toBe('Instant');
+ });
+});
diff --git a/connectors/grafana-plugin/src/types.ts
b/connectors/grafana-plugin/src/types.ts
index fb8a300..ae0d85e 100644
--- a/connectors/grafana-plugin/src/types.ts
+++ b/connectors/grafana-plugin/src/types.ts
@@ -41,6 +41,10 @@ export interface IoTDBQuery extends DataQuery {
sql?: string;
format?: string;
legendFormat?: string;
+ // Grafana query semantics. Queries saved before these fields existed are
+ // Range queries; Instant + Range together means Both.
+ instant?: boolean;
+ range?: boolean;
}
export interface GroupBy {
@@ -78,3 +82,24 @@ export interface IoTDBOptions extends DataSourceJsonData {
export interface IoTDBSecureJsonData {
password?: string;
}
+
+export const queryTypes = ['Range', 'Instant', 'Both'] as const;
+export type QueryType = (typeof queryTypes)[number];
+
+export function getQueryType(query: Pick<IoTDBQuery, 'instant' | 'range'>):
QueryType {
+ if (query.instant) {
+ return query.range ? 'Both' : 'Instant';
+ }
+ return 'Range';
+}
+
+export function applyQueryType(query: IoTDBQuery, queryType: QueryType):
IoTDBQuery {
+ switch (queryType) {
+ case 'Instant':
+ return { ...query, instant: true, range: false };
+ case 'Both':
+ return { ...query, instant: true, range: true };
+ default:
+ return { ...query, instant: false, range: true };
+ }
+}
