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 a811f62 Add backend expansion for Grafana $__interval and
$__interval_ms macros (#122)
a811f62 is described below
commit a811f627b7bced2cbfb1d142a7af4cf84d007dc2
Author: Zhao Xinqi <[email protected]>
AuthorDate: Wed Aug 12 16:10:47 2026 +0800
Add backend expansion for Grafana $__interval and $__interval_ms macros
(#122)
---
connectors/grafana-plugin/pkg/plugin/plugin.go | 10 +-
.../grafana-plugin/pkg/plugin/table_query.go | 108 +++++++++++++++-
.../grafana-plugin/pkg/plugin/table_query_test.go | 143 ++++++++++++++++++++-
3 files changed, 253 insertions(+), 8 deletions(-)
diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go
b/connectors/grafana-plugin/pkg/plugin/plugin.go
index 4b362ba..bc256cd 100644
--- a/connectors/grafana-plugin/pkg/plugin/plugin.go
+++ b/connectors/grafana-plugin/pkg/plugin/plugin.go
@@ -157,6 +157,7 @@ type queryParam struct {
Database string `json:"database"`
Sql string `json:"sql"`
Format string `json:"format"`
+ IntervalMS int64 `json:"-"`
}
type QueryDataReq struct {
@@ -262,8 +263,7 @@ func (d *IoTDBDataSource) query(cxt context.Context, pCtx
backend.PluginContext,
return response
}
- qp.StartTime = query.TimeRange.From.UnixNano() / 1000000
- qp.EndTime = query.TimeRange.To.UnixNano() / 1000000
+ applyQueryRuntimeValues(qp, query)
if qp.SqlType == TableModelSqlType {
return d.queryTableModel(cxt, qp)
@@ -358,6 +358,12 @@ func (d *IoTDBDataSource) query(cxt context.Context, pCtx
backend.PluginContext,
return response
}
+func applyQueryRuntimeValues(qp *queryParam, query backend.DataQuery) {
+ qp.StartTime = query.TimeRange.From.UnixNano() / 1000000
+ qp.EndTime = query.TimeRange.To.UnixNano() / 1000000
+ qp.IntervalMS = query.Interval.Milliseconds()
+}
+
func recoverType(m []interface{}) interface{} {
if len(m) > 0 {
switch m[0].(type) {
diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go
b/connectors/grafana-plugin/pkg/plugin/table_query.go
index a53eff1..7c115ae 100644
--- a/connectors/grafana-plugin/pkg/plugin/table_query.go
+++ b/connectors/grafana-plugin/pkg/plugin/table_query.go
@@ -73,8 +73,15 @@ var timeFilterRe =
regexp.MustCompile(`\$__timeFilter\(\s*((?:[^()]|\([^()]*\))*
var (
timeFromRe = regexp.MustCompile(`\$__timeFrom\b(?:\s*\(\s*\))?`)
timeToRe = regexp.MustCompile(`\$__timeTo\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.
+ intervalRe = regexp.MustCompile(`\$__interval`)
+ intervalMSRe = regexp.MustCompile(`\$__interval_ms`)
)
+const invalidIntervalMacroMessage = "Grafana query interval must be positive
when $__interval or $__interval_ms is used"
+
// 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
@@ -84,16 +91,37 @@ 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:
+// expandTableMacros rewrites the Grafana time and interval macros a dashboard
+// author can put in table-model SQL.
//
// $__timeFilter(col) -> (col >= <from> AND col <= <to>)
// $__timeFrom[()] -> <from>
// $__timeTo[()] -> <to>
+// $__interval -> a fixed-width IoTDB duration literal
+// $__interval_ms -> the interval in milliseconds, per Grafana's
contract
//
// 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 {
+func expandTableMacros(sql string, startMs int64, endMs int64, intervalMS
int64) (string, error) {
+ hasInterval := hasStandaloneMacro(sql, intervalRe)
+ hasIntervalMS := hasStandaloneMacro(sql, intervalMSRe)
+ if (hasInterval || hasIntervalMS) && intervalMS <= 0 {
+ // Defensive validation for direct callers. queryTableModel
performs the
+ // authoritative request-path check before acquiring an RPC
session.
+ return "", errors.New(invalidIntervalMacroMessage)
+ }
+
+ if hasIntervalMS {
+ sql = replaceStandaloneMacro(sql, intervalMSRe,
strconv.FormatInt(intervalMS, 10))
+ }
+ if hasInterval {
+ duration, err := formatIoTDBDuration(intervalMS)
+ if err != nil {
+ return "", err
+ }
+ sql = replaceStandaloneMacro(sql, intervalRe, duration)
+ }
+
from := formatTimeLiteral(startMs)
to := formatTimeLiteral(endMs)
sql = timeFilterRe.ReplaceAllStringFunc(sql, func(m string) string {
@@ -105,7 +133,67 @@ func expandTableMacros(sql string, startMs int64, endMs
int64) string {
})
sql = timeFromRe.ReplaceAllString(sql, from)
sql = timeToRe.ReplaceAllString(sql, to)
- return sql
+ return sql, nil
+}
+
+// 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.
+func hasStandaloneMacro(sql string, re *regexp.Regexp) bool {
+ for _, match := range re.FindAllStringIndex(sql, -1) {
+ if match[1] == len(sql) || !isSQLIdentifierByte(sql[match[1]]) {
+ return true
+ }
+ }
+ return false
+}
+
+func replaceStandaloneMacro(sql string, re *regexp.Regexp, replacement string)
string {
+ matches := re.FindAllStringIndex(sql, -1)
+ if len(matches) == 0 {
+ return sql
+ }
+ var b strings.Builder
+ last := 0
+ for _, match := range matches {
+ if match[1] < len(sql) && isSQLIdentifierByte(sql[match[1]]) {
+ continue
+ }
+ b.WriteString(sql[last:match[0]])
+ b.WriteString(replacement)
+ last = match[1]
+ }
+ b.WriteString(sql[last:])
+ return b.String()
+}
+
+func isSQLIdentifierByte(b byte) bool {
+ return b == '_' || b >= 'a' && b <= 'z' || b >= 'A' && b <= 'Z' || b >=
'0' && b <= '9'
+}
+
+// formatIoTDBDuration uses only fixed-width units accepted by IoTDB and
+// avoids calendar month/year semantics. The largest exact unit is selected.
+func formatIoTDBDuration(intervalMS int64) (string, error) {
+ if intervalMS <= 0 {
+ return "", errors.New("Grafana query interval must be positive")
+ }
+ units := []struct {
+ milliseconds int64
+ suffix string
+ }{
+ {7 * 24 * 60 * 60 * 1000, "w"},
+ {24 * 60 * 60 * 1000, "d"},
+ {60 * 60 * 1000, "h"},
+ {60 * 1000, "m"},
+ {1000, "s"},
+ {1, "ms"},
+ }
+ for _, unit := range units {
+ if intervalMS%unit.milliseconds == 0 {
+ return strconv.FormatInt(intervalMS/unit.milliseconds,
10) + unit.suffix, nil
+ }
+ }
+ return "", errors.New("cannot format Grafana query interval")
}
// quoteTableIdentifier wraps a table-model identifier in double quotes
@@ -215,7 +303,12 @@ func (d *IoTDBDataSource) getTablePool()
(*client.TableSessionPool, error) {
func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam)
backend.DataResponse {
response := backend.DataResponse{}
- sql := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime)
+ if (hasStandaloneMacro(qp.Sql, intervalRe) ||
hasStandaloneMacro(qp.Sql, intervalMSRe)) && qp.IntervalMS <= 0 {
+ // This is the authoritative guard: reject invalid Grafana
input before
+ // getTablePool can create or acquire an RPC session.
+ response.Error = errors.New(invalidIntervalMacroMessage)
+ return response
+ }
pool, err := d.getTablePool()
if err != nil {
@@ -247,6 +340,11 @@ func (d *IoTDBDataSource) queryTableModel(ctx
context.Context, qp *queryParam) b
timeout = ms
}
}
+ sql, err := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime,
qp.IntervalMS)
+ if err != nil {
+ response.Error = err
+ return response
+ }
resultSet, err := session.ExecuteQueryStatement(sql, &timeout)
if err != nil {
response.Error = err
diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go
b/connectors/grafana-plugin/pkg/plugin/table_query_test.go
index 899bc14..29c65eb 100644
--- a/connectors/grafana-plugin/pkg/plugin/table_query_test.go
+++ b/connectors/grafana-plugin/pkg/plugin/table_query_test.go
@@ -18,7 +18,10 @@
package plugin
import (
+ "context"
+ "encoding/json"
"errors"
+ "strings"
"testing"
"time"
@@ -80,7 +83,10 @@ func TestExpandTableMacros(t *testing.T) {
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
- got := expandTableMacros(c.in, from, to)
+ got, err := expandTableMacros(c.in, from, to, 0)
+ if err != nil {
+ t.Fatalf("expandTableMacros() unexpected error:
%v", err)
+ }
if got != c.want {
t.Fatalf("expandTableMacros() = %q, want %q",
got, c.want)
}
@@ -88,6 +94,141 @@ func TestExpandTableMacros(t *testing.T) {
}
}
+func TestExpandTableIntervalMacros(t *testing.T) {
+ const from int64 = 1600000000000
+ const to int64 = 1600000001000
+
+ cases := []struct {
+ name string
+ sql string
+ interval int64
+ want string
+ wantErr string
+ }{
+ {
+ name: "expands duration and milliseconds together",
+ sql: "SELECT date_bin($__interval, time) +
$__interval_ms AS bucket_time FROM table1",
+ interval: 120000,
+ want: "SELECT date_bin(2m, time) + 120000 AS
bucket_time FROM table1",
+ },
+ {name: "separate interval macros do not overlap", sql: "SELECT
$__interval, $__interval_ms", interval: 120000, want: "SELECT 2m, 120000"},
+ {name: "milliseconds", sql: "SELECT $__interval", interval:
500, want: "SELECT 500ms"},
+ {name: "seconds", sql: "SELECT $__interval", interval: 1000,
want: "SELECT 1s"},
+ {name: "minutes", sql: "SELECT $__interval", interval: 120000,
want: "SELECT 2m"},
+ {name: "hours", sql: "SELECT $__interval", interval: 3600000,
want: "SELECT 1h"},
+ {name: "days", sql: "SELECT $__interval", interval: 86400000,
want: "SELECT 1d"},
+ {name: "weeks", sql: "SELECT $__interval", interval: 604800000,
want: "SELECT 1w"},
+ {name: "non exact duration uses milliseconds", sql: "SELECT
$__interval", interval: 1500, want: "SELECT 1500ms"},
+ {name: "Grafana interval milliseconds contract", sql: "SELECT
$__interval_ms", interval: 120000, want: "SELECT 120000"},
+ {name: "identifier boundaries are preserved", sql: "SELECT
$__intervalish, $__interval_ms_extra", interval: 120000, want: "SELECT
$__intervalish, $__interval_ms_extra"},
+ {name: "interval is ignored when no interval macro exists",
sql: "SELECT $__timeFrom", interval: 0, want: "SELECT
2020-09-13T12:26:40.000+00:00"},
+ {name: "zero interval fails", sql: "SELECT $__interval",
interval: 0, wantErr: "Grafana query interval must be positive"},
+ {name: "negative interval fails", sql: "SELECT $__interval_ms",
interval: -1, wantErr: "Grafana query interval must be positive"},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ got, err := expandTableMacros(tc.sql, from, to,
tc.interval)
+ if tc.wantErr != "" {
+ if err == nil || !strings.Contains(err.Error(),
tc.wantErr) {
+ t.Fatalf("expandTableMacros() error =
%v, want substring %q", err, tc.wantErr)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("expandTableMacros() unexpected error:
%v", err)
+ }
+ if got != tc.want {
+ t.Fatalf("expandTableMacros() = %q, want %q",
got, tc.want)
+ }
+ })
+ }
+}
+
+func TestExpandTableIntervalMacrosWithExplicitOrigin(t *testing.T) {
+ cases := []struct {
+ name string
+ from int64
+ to int64
+ sql string
+ want string
+ }{
+ {
+ name: "date bin origin offset by thirty seconds",
+ from: 1600000030000,
+ to: 1600000150000,
+ sql: "SELECT date_bin($__interval, time, $__timeFrom)
AS bucket_time FROM table1",
+ want: "SELECT date_bin(2m, time,
2020-09-13T12:27:10.000+00:00) AS bucket_time FROM table1",
+ },
+ {
+ name: "hop origin offset by forty five seconds",
+ from: 1600000045000,
+ to: 1600000165000,
+ sql: "SELECT * FROM HOP(DATA => table1, SLIDE =>
$__interval, SIZE => 1m, ORIGIN => $__timeFrom)",
+ want: "SELECT * FROM HOP(DATA => table1, SLIDE => 2m,
SIZE => 1m, ORIGIN => 2020-09-13T12:27:25.000+00:00)",
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ got, err := expandTableMacros(tc.sql, tc.from, tc.to,
120000)
+ if err != nil {
+ t.Fatalf("expandTableMacros() unexpected error:
%v", err)
+ }
+ if got != tc.want {
+ t.Fatalf("expandTableMacros() = %q, want %q",
got, tc.want)
+ }
+ })
+ }
+}
+
+func TestQueryParamRuntimeIntervalOverridesJSON(t *testing.T) {
+ qp, msg := verifyQuery(backend.DataQuery{JSON: []byte(`{"sqlType":"SQL:
Table Model","sql":"SELECT
$__interval_ms","database":"db1","intervalMS":999}`)})
+ if msg != "" {
+ t.Fatalf("valid table query rejected: %q", msg)
+ }
+ if qp.IntervalMS != 0 {
+ t.Fatalf("JSON intervalMS should not populate runtime field:
%d", qp.IntervalMS)
+ }
+ serialized, err := json.Marshal(qp)
+ if err != nil {
+ t.Fatalf("marshal query param: %v", err)
+ }
+ if strings.Contains(string(serialized), "intervalMS") {
+ t.Fatalf("runtime intervalMS must not be serialized: %s",
serialized)
+ }
+ query := backend.DataQuery{
+ Interval: 120 * time.Second,
+ TimeRange: backend.TimeRange{
+ From: ts(1600000000000),
+ To: ts(1600000001000),
+ },
+ }
+ applyQueryRuntimeValues(qp, query)
+ if qp.IntervalMS != 120000 || qp.StartTime != 1600000000000 ||
qp.EndTime != 1600000001000 {
+ t.Fatalf("runtime values = start %d, end %d, interval %d",
qp.StartTime, qp.EndTime, qp.IntervalMS)
+ }
+}
+
+func TestQueryTableModelRejectsNonPositiveIntervalBeforeRPC(t *testing.T) {
+ _, expandErr := expandTableMacros("SELECT $__interval", 0, 0, 0)
+ if expandErr == nil || expandErr.Error() != invalidIntervalMacroMessage
{
+ t.Fatalf("expandTableMacros() error = %v, want %q", expandErr,
invalidIntervalMacroMessage)
+ }
+
+ d := &IoTDBDataSource{Ulr: "http://invalid-host:18080"}
+ response := d.queryTableModel(context.Background(), &queryParam{
+ Sql: "SELECT $__interval FROM table1",
+ Database: "db1",
+ IntervalMS: 0,
+ })
+ if response.Error == nil || response.Error.Error() !=
invalidIntervalMacroMessage {
+ t.Fatalf("queryTableModel() error = %v, want early
positive-interval error", response.Error)
+ }
+ if d.tablePool != nil {
+ t.Fatalf("invalid interval should be rejected before creating
an RPC pool")
+ }
+}
+
func TestQuoteTableIdentifier(t *testing.T) {
if got := quoteTableIdentifier("test"); got != `"test"` {
t.Fatalf("plain identifier = %q", got)
