Add backend expansion for Grafana $__interval and $__interval_ms macros

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) {

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

connectors/grafana-plugin/pkg/plugin/table_query_test.go

@@ -18,7 +18,10 @@
 package plugin
 
 import (
+	"context"
+	"encoding/json"
 	"errors"
+	"strings"
 	"testing"
 	"time"
 
@@ -80,14 +83,152 @@ 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)
 			}
 		})
 	}
 }
 
+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)