From 4763988bccf08a3b340f093ea77736af03ad6564 Mon Sep 17 00:00:00 2001 From: zxq <3322351820@qq.com> Date: Tue, 11 Aug 2026 10:10:38 +0800 Subject: [PATCH 1/2] Add backend expansion for Grafana $__interval and $__interval_ms macros. --- .../grafana-plugin/pkg/plugin/plugin.go | 10 +- .../grafana-plugin/pkg/plugin/table_query.go | 208 +++++++++++++++++- .../pkg/plugin/table_query_test.go | 201 ++++++++++++++++- 3 files changed, 411 insertions(+), 8 deletions(-) diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go b/connectors/grafana-plugin/pkg/plugin/plugin.go index 4b362ba..a495ffd 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..142ee7f 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,51 @@ 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. The precision-aware RPC path calls the +// internal helper with the server's timestamp precision. // // $__timeFilter(col) -> (col >= AND col <= ) // $__timeFrom[()] -> // $__timeTo[()] -> +// $__interval -> a fixed-width IoTDB duration literal +// $__interval_ms -> the interval in server timestamp units // // 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) { + return expandTableMacrosWithPrecision(sql, startMs, endMs, intervalMS, "ms") +} + +// expandTableMacrosWithPrecision expands the two Grafana interval macros in +// addition to the existing time macros. intervalMS is Grafana's runtime +// suggested step in milliseconds; it is never read from Dashboard JSON. +// timestampPrecision controls the unit of integer TIMESTAMP arithmetic used by +// $__interval_ms and must be ms, us, or ns. +func expandTableMacrosWithPrecision(sql string, startMs int64, endMs int64, intervalMS int64, timestampPrecision string) (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 { + scaled, err := scaleIntervalMS(intervalMS, timestampPrecision) + if err != nil { + return "", err + } + sql = replaceStandaloneMacro(sql, intervalMSRe, strconv.FormatInt(scaled, 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 +147,89 @@ 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") +} + +func scaleIntervalMS(intervalMS int64, timestampPrecision string) (int64, error) { + if intervalMS <= 0 { + return 0, errors.New("Grafana query interval must be positive") + } + var multiplier int64 + switch strings.ToLower(strings.TrimSpace(timestampPrecision)) { + case "ms": + multiplier = 1 + case "us": + multiplier = 1000 + case "ns": + multiplier = 1000000 + default: + return 0, fmt.Errorf("unsupported IoTDB timestamp precision %q", timestampPrecision) + } + maxInt64 := int64(^uint64(0) >> 1) + if intervalMS > maxInt64/multiplier { + return 0, fmt.Errorf("Grafana query interval %dms overflows IoTDB %s timestamp units", intervalMS, timestampPrecision) + } + return intervalMS * multiplier, nil } // quoteTableIdentifier wraps a table-model identifier in double quotes @@ -215,7 +339,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 +376,19 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b timeout = ms } } + timestampPrecision := "ms" + if hasStandaloneMacro(qp.Sql, intervalMSRe) { + timestampPrecision, err = readTimestampPrecision(session, &timeout) + if err != nil { + response.Error = fmt.Errorf("cannot determine IoTDB timestamp precision for $__interval_ms: %w", err) + return response + } + } + sql, err := expandTableMacrosWithPrecision(qp.Sql, qp.StartTime, qp.EndTime, qp.IntervalMS, timestampPrecision) + if err != nil { + response.Error = err + return response + } resultSet, err := session.ExecuteQueryStatement(sql, &timeout) if err != nil { response.Error = err @@ -268,6 +410,62 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b return response } +func readTimestampPrecision(session client.ITableSession, timeout *int64) (string, error) { + resultSet, err := session.ExecuteQueryStatement("SHOW VARIABLES", timeout) + if err != nil { + return "", err + } + defer resultSet.Close() + dataSet, err := fetchTableDataSet(resultSet) + if err != nil { + return "", err + } + return timestampPrecisionFromDataSet(dataSet) +} + +func timestampPrecisionFromDataSet(dataSet *tableQueryDataSet) (string, error) { + columnNames := dataSet.ColumnNames + variableIndex, valueIndex := -1, -1 + for i, name := range columnNames { + switch strings.ToLower(strings.TrimSpace(name)) { + case "variable": + variableIndex = i + case "value": + valueIndex = i + } + } + if variableIndex < 0 || valueIndex < 0 { + return "", fmt.Errorf("SHOW VARIABLES result does not contain Variable and Value columns") + } + for _, row := range dataSet.Values { + if variableIndex >= len(row) || valueIndex >= len(row) { + continue + } + variable := row[variableIndex] + if !strings.EqualFold(strings.TrimSpace(sqlScalarString(variable)), "TimestampPrecision") { + continue + } + value := row[valueIndex] + precision := strings.ToLower(strings.TrimSpace(sqlScalarString(value))) + if precision != "ms" && precision != "us" && precision != "ns" { + return "", fmt.Errorf("unsupported IoTDB timestamp precision %q", precision) + } + return precision, nil + } + return "", errors.New("SHOW VARIABLES did not return TimestampPrecision") +} + +func sqlScalarString(value interface{}) string { + switch v := value.(type) { + case string: + return v + case []byte: + return string(v) + default: + return fmt.Sprint(v) + } +} + // 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 diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go b/connectors/grafana-plugin/pkg/plugin/table_query_test.go index 899bc14..939410e 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,151 @@ 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 + precision string + want string + wantErr string + }{ + { + name: "expands duration and raw milliseconds together", + sql: "SELECT date_bin($__interval, time) + $__interval_ms AS bucket_time FROM table1", + interval: 120000, + precision: "ms", + 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, + precision: "ms", + want: "SELECT 2m, 120000", + }, + {name: "milliseconds", sql: "SELECT $__interval", interval: 500, precision: "ms", want: "SELECT 500ms"}, + {name: "seconds", sql: "SELECT $__interval", interval: 1000, precision: "ms", want: "SELECT 1s"}, + {name: "minutes", sql: "SELECT $__interval", interval: 120000, precision: "ms", want: "SELECT 2m"}, + {name: "hours", sql: "SELECT $__interval", interval: 3600000, precision: "ms", want: "SELECT 1h"}, + {name: "days", sql: "SELECT $__interval", interval: 86400000, precision: "ms", want: "SELECT 1d"}, + {name: "weeks", sql: "SELECT $__interval", interval: 604800000, precision: "ms", want: "SELECT 1w"}, + {name: "non exact duration uses milliseconds", sql: "SELECT $__interval", interval: 1500, precision: "ms", want: "SELECT 1500ms"}, + {name: "microsecond server scales raw timestamp arithmetic", sql: "SELECT $__interval_ms", interval: 120000, precision: "us", want: "SELECT 120000000"}, + {name: "nanosecond server scales raw timestamp arithmetic", sql: "SELECT $__interval_ms", interval: 120000, precision: "ns", want: "SELECT 120000000000"}, + {name: "identifier boundaries are preserved", sql: "SELECT $__intervalish, $__interval_ms_extra", interval: 120000, precision: "ms", want: "SELECT $__intervalish, $__interval_ms_extra"}, + {name: "interval is ignored when no interval macro exists", sql: "SELECT $__timeFrom", interval: 0, precision: "ms", want: "SELECT 2020-09-13T12:26:40.000+00:00"}, + {name: "zero interval fails", sql: "SELECT $__interval", interval: 0, precision: "ms", wantErr: "Grafana query interval must be positive"}, + {name: "negative interval fails", sql: "SELECT $__interval_ms", interval: -1, precision: "ms", wantErr: "Grafana query interval must be positive"}, + {name: "unknown precision fails", sql: "SELECT $__interval_ms", interval: 1000, precision: "ps", wantErr: "unsupported IoTDB timestamp precision"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := expandTableMacrosWithPrecision(tc.sql, from, to, tc.interval, tc.precision) + 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) @@ -162,6 +313,54 @@ func TestFetchTableDataSetPropagatesError(t *testing.T) { } } +func TestTimestampPrecisionFromDataSet(t *testing.T) { + cases := []struct { + name string + dataSet *tableQueryDataSet + want string + errText string + }{ + { + name: "case insensitive variable and value columns", + dataSet: &tableQueryDataSet{ + ColumnNames: []string{"Value", "Variable"}, + Values: [][]interface{}{{[]byte("ms"), []byte("TimestampPrecision")}}, + }, + want: "ms", + }, + { + name: "unsupported precision", + dataSet: &tableQueryDataSet{ + ColumnNames: []string{"Variable", "Value"}, + Values: [][]interface{}{{"TimestampPrecision", "ps"}}, + }, + errText: "unsupported IoTDB timestamp precision", + }, + { + name: "missing row", + dataSet: &tableQueryDataSet{ + ColumnNames: []string{"Variable", "Value"}, + Values: [][]interface{}{{"ClusterName", "defaultCluster"}}, + }, + errText: "did not return TimestampPrecision", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := timestampPrecisionFromDataSet(tc.dataSet) + if tc.errText != "" { + if err == nil || !strings.Contains(err.Error(), tc.errText) { + t.Fatalf("timestampPrecisionFromDataSet() error = %v, want substring %q", err, tc.errText) + } + return + } + if err != nil || got != tc.want { + t.Fatalf("timestampPrecisionFromDataSet() = %q, %v; want %q", got, err, tc.want) + } + }) + } +} + // 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. From 1199c74e2d80c471ddbedf03341c3789966b0c22 Mon Sep 17 00:00:00 2001 From: zxq <3322351820@qq.com> Date: Wed, 12 Aug 2026 14:28:12 +0800 Subject: [PATCH 2/2] expand $__interval_ms as raw milliseconds and drop SHOW VARIABLES lookup --- .../grafana-plugin/pkg/plugin/plugin.go | 2 +- .../grafana-plugin/pkg/plugin/table_query.go | 108 +----------------- .../pkg/plugin/table_query_test.go | 104 ++++------------- 3 files changed, 28 insertions(+), 186 deletions(-) diff --git a/connectors/grafana-plugin/pkg/plugin/plugin.go b/connectors/grafana-plugin/pkg/plugin/plugin.go index a495ffd..bc256cd 100644 --- a/connectors/grafana-plugin/pkg/plugin/plugin.go +++ b/connectors/grafana-plugin/pkg/plugin/plugin.go @@ -157,7 +157,7 @@ type queryParam struct { Database string `json:"database"` Sql string `json:"sql"` Format string `json:"format"` - IntervalMS int64 `json:"-"` + IntervalMS int64 `json:"-"` } type QueryDataReq struct { diff --git a/connectors/grafana-plugin/pkg/plugin/table_query.go b/connectors/grafana-plugin/pkg/plugin/table_query.go index 142ee7f..7c115ae 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_query.go +++ b/connectors/grafana-plugin/pkg/plugin/table_query.go @@ -92,27 +92,17 @@ func formatTimeLiteral(ms int64) string { } // expandTableMacros rewrites the Grafana time and interval macros a dashboard -// author can put in table-model SQL. The precision-aware RPC path calls the -// internal helper with the server's timestamp precision. +// author can put in table-model SQL. // // $__timeFilter(col) -> (col >= AND col <= ) // $__timeFrom[()] -> // $__timeTo[()] -> // $__interval -> a fixed-width IoTDB duration literal -// $__interval_ms -> the interval in server timestamp units +// $__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, intervalMS int64) (string, error) { - return expandTableMacrosWithPrecision(sql, startMs, endMs, intervalMS, "ms") -} - -// expandTableMacrosWithPrecision expands the two Grafana interval macros in -// addition to the existing time macros. intervalMS is Grafana's runtime -// suggested step in milliseconds; it is never read from Dashboard JSON. -// timestampPrecision controls the unit of integer TIMESTAMP arithmetic used by -// $__interval_ms and must be ms, us, or ns. -func expandTableMacrosWithPrecision(sql string, startMs int64, endMs int64, intervalMS int64, timestampPrecision string) (string, error) { hasInterval := hasStandaloneMacro(sql, intervalRe) hasIntervalMS := hasStandaloneMacro(sql, intervalMSRe) if (hasInterval || hasIntervalMS) && intervalMS <= 0 { @@ -122,11 +112,7 @@ func expandTableMacrosWithPrecision(sql string, startMs int64, endMs int64, inte } if hasIntervalMS { - scaled, err := scaleIntervalMS(intervalMS, timestampPrecision) - if err != nil { - return "", err - } - sql = replaceStandaloneMacro(sql, intervalMSRe, strconv.FormatInt(scaled, 10)) + sql = replaceStandaloneMacro(sql, intervalMSRe, strconv.FormatInt(intervalMS, 10)) } if hasInterval { duration, err := formatIoTDBDuration(intervalMS) @@ -210,28 +196,6 @@ func formatIoTDBDuration(intervalMS int64) (string, error) { return "", errors.New("cannot format Grafana query interval") } -func scaleIntervalMS(intervalMS int64, timestampPrecision string) (int64, error) { - if intervalMS <= 0 { - return 0, errors.New("Grafana query interval must be positive") - } - var multiplier int64 - switch strings.ToLower(strings.TrimSpace(timestampPrecision)) { - case "ms": - multiplier = 1 - case "us": - multiplier = 1000 - case "ns": - multiplier = 1000000 - default: - return 0, fmt.Errorf("unsupported IoTDB timestamp precision %q", timestampPrecision) - } - maxInt64 := int64(^uint64(0) >> 1) - if intervalMS > maxInt64/multiplier { - return 0, fmt.Errorf("Grafana query interval %dms overflows IoTDB %s timestamp units", intervalMS, timestampPrecision) - } - return intervalMS * multiplier, nil -} - // 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. @@ -376,15 +340,7 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b timeout = ms } } - timestampPrecision := "ms" - if hasStandaloneMacro(qp.Sql, intervalMSRe) { - timestampPrecision, err = readTimestampPrecision(session, &timeout) - if err != nil { - response.Error = fmt.Errorf("cannot determine IoTDB timestamp precision for $__interval_ms: %w", err) - return response - } - } - sql, err := expandTableMacrosWithPrecision(qp.Sql, qp.StartTime, qp.EndTime, qp.IntervalMS, timestampPrecision) + sql, err := expandTableMacros(qp.Sql, qp.StartTime, qp.EndTime, qp.IntervalMS) if err != nil { response.Error = err return response @@ -410,62 +366,6 @@ func (d *IoTDBDataSource) queryTableModel(ctx context.Context, qp *queryParam) b return response } -func readTimestampPrecision(session client.ITableSession, timeout *int64) (string, error) { - resultSet, err := session.ExecuteQueryStatement("SHOW VARIABLES", timeout) - if err != nil { - return "", err - } - defer resultSet.Close() - dataSet, err := fetchTableDataSet(resultSet) - if err != nil { - return "", err - } - return timestampPrecisionFromDataSet(dataSet) -} - -func timestampPrecisionFromDataSet(dataSet *tableQueryDataSet) (string, error) { - columnNames := dataSet.ColumnNames - variableIndex, valueIndex := -1, -1 - for i, name := range columnNames { - switch strings.ToLower(strings.TrimSpace(name)) { - case "variable": - variableIndex = i - case "value": - valueIndex = i - } - } - if variableIndex < 0 || valueIndex < 0 { - return "", fmt.Errorf("SHOW VARIABLES result does not contain Variable and Value columns") - } - for _, row := range dataSet.Values { - if variableIndex >= len(row) || valueIndex >= len(row) { - continue - } - variable := row[variableIndex] - if !strings.EqualFold(strings.TrimSpace(sqlScalarString(variable)), "TimestampPrecision") { - continue - } - value := row[valueIndex] - precision := strings.ToLower(strings.TrimSpace(sqlScalarString(value))) - if precision != "ms" && precision != "us" && precision != "ns" { - return "", fmt.Errorf("unsupported IoTDB timestamp precision %q", precision) - } - return precision, nil - } - return "", errors.New("SHOW VARIABLES did not return TimestampPrecision") -} - -func sqlScalarString(value interface{}) string { - switch v := value.(type) { - case string: - return v - case []byte: - return string(v) - default: - return fmt.Sprint(v) - } -} - // 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 diff --git a/connectors/grafana-plugin/pkg/plugin/table_query_test.go b/connectors/grafana-plugin/pkg/plugin/table_query_test.go index 939410e..29c65eb 100644 --- a/connectors/grafana-plugin/pkg/plugin/table_query_test.go +++ b/connectors/grafana-plugin/pkg/plugin/table_query_test.go @@ -99,46 +99,36 @@ func TestExpandTableIntervalMacros(t *testing.T) { const to int64 = 1600000001000 cases := []struct { - name string - sql string - interval int64 - precision string - want string - wantErr string + name string + sql string + interval int64 + want string + wantErr string }{ { - name: "expands duration and raw milliseconds together", - sql: "SELECT date_bin($__interval, time) + $__interval_ms AS bucket_time FROM table1", - interval: 120000, - precision: "ms", - want: "SELECT date_bin(2m, time) + 120000 AS bucket_time FROM table1", + 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, - precision: "ms", - want: "SELECT 2m, 120000", - }, - {name: "milliseconds", sql: "SELECT $__interval", interval: 500, precision: "ms", want: "SELECT 500ms"}, - {name: "seconds", sql: "SELECT $__interval", interval: 1000, precision: "ms", want: "SELECT 1s"}, - {name: "minutes", sql: "SELECT $__interval", interval: 120000, precision: "ms", want: "SELECT 2m"}, - {name: "hours", sql: "SELECT $__interval", interval: 3600000, precision: "ms", want: "SELECT 1h"}, - {name: "days", sql: "SELECT $__interval", interval: 86400000, precision: "ms", want: "SELECT 1d"}, - {name: "weeks", sql: "SELECT $__interval", interval: 604800000, precision: "ms", want: "SELECT 1w"}, - {name: "non exact duration uses milliseconds", sql: "SELECT $__interval", interval: 1500, precision: "ms", want: "SELECT 1500ms"}, - {name: "microsecond server scales raw timestamp arithmetic", sql: "SELECT $__interval_ms", interval: 120000, precision: "us", want: "SELECT 120000000"}, - {name: "nanosecond server scales raw timestamp arithmetic", sql: "SELECT $__interval_ms", interval: 120000, precision: "ns", want: "SELECT 120000000000"}, - {name: "identifier boundaries are preserved", sql: "SELECT $__intervalish, $__interval_ms_extra", interval: 120000, precision: "ms", want: "SELECT $__intervalish, $__interval_ms_extra"}, - {name: "interval is ignored when no interval macro exists", sql: "SELECT $__timeFrom", interval: 0, precision: "ms", want: "SELECT 2020-09-13T12:26:40.000+00:00"}, - {name: "zero interval fails", sql: "SELECT $__interval", interval: 0, precision: "ms", wantErr: "Grafana query interval must be positive"}, - {name: "negative interval fails", sql: "SELECT $__interval_ms", interval: -1, precision: "ms", wantErr: "Grafana query interval must be positive"}, - {name: "unknown precision fails", sql: "SELECT $__interval_ms", interval: 1000, precision: "ps", wantErr: "unsupported IoTDB timestamp precision"}, + {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 := expandTableMacrosWithPrecision(tc.sql, from, to, tc.interval, tc.precision) + 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) @@ -313,54 +303,6 @@ func TestFetchTableDataSetPropagatesError(t *testing.T) { } } -func TestTimestampPrecisionFromDataSet(t *testing.T) { - cases := []struct { - name string - dataSet *tableQueryDataSet - want string - errText string - }{ - { - name: "case insensitive variable and value columns", - dataSet: &tableQueryDataSet{ - ColumnNames: []string{"Value", "Variable"}, - Values: [][]interface{}{{[]byte("ms"), []byte("TimestampPrecision")}}, - }, - want: "ms", - }, - { - name: "unsupported precision", - dataSet: &tableQueryDataSet{ - ColumnNames: []string{"Variable", "Value"}, - Values: [][]interface{}{{"TimestampPrecision", "ps"}}, - }, - errText: "unsupported IoTDB timestamp precision", - }, - { - name: "missing row", - dataSet: &tableQueryDataSet{ - ColumnNames: []string{"Variable", "Value"}, - Values: [][]interface{}{{"ClusterName", "defaultCluster"}}, - }, - errText: "did not return TimestampPrecision", - }, - } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - got, err := timestampPrecisionFromDataSet(tc.dataSet) - if tc.errText != "" { - if err == nil || !strings.Contains(err.Error(), tc.errText) { - t.Fatalf("timestampPrecisionFromDataSet() error = %v, want substring %q", err, tc.errText) - } - return - } - if err != nil || got != tc.want { - t.Fatalf("timestampPrecisionFromDataSet() = %q, %v; want %q", got, err, tc.want) - } - }) - } -} - // 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.