public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SQL data sources: Convert to return data frames (#32257)

Browse Source 
Convert SQL data sources to return data frames.

Co-authored-by: Marcus Efraimsson <marcus.efraimsson@gmail.com>
Co-authored-by: Arve Knudsen <arve.knudsen@gmail.com>
Co-authored-by: Will Browne <will.browne@grafana.com>
Co-authored-by: Hugo Häggmark <hugo.haggmark@gmail.com>
This commit is contained in:
ying-jeanne
2021-05-05 16:46:07 +02:00
committed by GitHub
co-authored by Marcus Efraimsson Arve Knudsen Will Browne Hugo Häggmark
parent 06c24476dc
commit bd66c8dde3
36 changed files with 4947 additions and 3711 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/sqlstore/sqlstore.go
+8
View File
@@ -510,6 +510,14 @@ func IsTestDbPostgres() bool {
return false
}
func IsTestDBMSSQL() bool {
if db, present := os.LookupEnv("GRAFANA_TEST_DB"); present {
return db == migrator.MSSQL
}
return false
}
type DatabaseConfig struct {
Type string
Host string
pkg/tsdb/interval/interval.go
+1 -1
View File
@@ -69,7 +69,7 @@ func (ic *intervalCalculator) Calculate(timerange plugins.DataTimeRange, minInte
func GetIntervalFrom(dsInfo *models.DataSource, queryModel *simplejson.Json, defaultInterval time.Duration) (time.Duration, error) {
interval := queryModel.Get("interval").MustString("")
if interval == "" && dsInfo.JsonData != nil {
if interval == "" && dsInfo != nil && dsInfo.JsonData != nil {
dsInterval := dsInfo.JsonData.Get("timeInterval").MustString("")
if dsInterval != "" {
interval = dsInterval
pkg/tsdb/mssql/mssql.go
+85 -45
View File
@@ -1,13 +1,15 @@
package mssql
import (
"database/sql"
"fmt"
"net/url"
"reflect"
"regexp"
"strconv"
"strings"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/util"
@@ -16,7 +18,6 @@ import (
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/tsdb/sqleng"
"xorm.io/core"
)
var logger = log.New("tsdb.mssql")
@@ -115,49 +116,6 @@ type mssqlQueryResultTransformer struct {
log log.Logger
}
func (t *mssqlQueryResultTransformer) TransformQueryResult(columnTypes []*sql.ColumnType, rows *core.Rows) (
plugins.DataRowValues, error) {
values := make([]interface{}, len(columnTypes))
valuePtrs := make([]interface{}, len(columnTypes))
for i := range columnTypes {
// debug output on large tables causes high memory utilization/leak
// t.log.Debug("type", "type", stype)
valuePtrs[i] = &values[i]
}
if err := rows.Scan(valuePtrs...); err != nil {
return nil, err
}
// convert types not handled by denisenkom/go-mssqldb
// unhandled types are returned as []byte
for i := 0; i < len(columnTypes); i++ {
if value, ok := values[i].([]byte); ok {
switch columnTypes[i].DatabaseTypeName() {
case "MONEY", "SMALLMONEY", "DECIMAL":
if v, err := strconv.ParseFloat(string(value), 64); err == nil {
values[i] = v
} else {
t.log.Debug("Rows", "Error converting numeric to float", value)
}
case "UNIQUEIDENTIFIER":
uuid := &mssql.UniqueIdentifier{}
if err := uuid.Scan(value); err == nil {
values[i] = uuid.String()
} else {
t.log.Debug("Rows", "Error converting uniqueidentifier to string", value)
}
default:
t.log.Debug("Rows", "Unknown database type", columnTypes[i].DatabaseTypeName(), "value", value)
values[i] = string(value)
}
}
}
return values, nil
}
func (t *mssqlQueryResultTransformer) TransformQueryError(err error) error {
// go-mssql overrides source error, so we currently match on string
// ref https://github.com/denisenkom/go-mssqldb/blob/045585d74f9069afe2e115b6235eb043c8047043/tds.go#L904
@@ -168,3 +126,85 @@ func (t *mssqlQueryResultTransformer) TransformQueryError(err error) error {
return err
}
func (t *mssqlQueryResultTransformer) GetConverterList() []sqlutil.StringConverter {
return []sqlutil.StringConverter{
{
Name: "handle MONEY",
InputScanKind: reflect.Slice,
InputTypeName: "MONEY",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle SMALLMONEY",
InputScanKind: reflect.Slice,
InputTypeName: "SMALLMONEY",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle DECIMAL",
InputScanKind: reflect.Slice,
InputTypeName: "DECIMAL",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle UNIQUEIDENTIFIER",
InputScanKind: reflect.Slice,
InputTypeName: "UNIQUEIDENTIFIER",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableString,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
uuid := &mssql.UniqueIdentifier{}
if err := uuid.Scan([]byte(*in)); err != nil {
return nil, err
}
v := uuid.String()
return &v, nil
},
},
},
}
}
pkg/tsdb/mssql/mssql_test.go
+1029 -967
View File
File diff suppressed because it is too large Load Diff
pkg/tsdb/mysql/mysql.go
+205 -62
View File
@@ -1,15 +1,17 @@
package mysql
import (
"database/sql"
"errors"
"fmt"
"net/url"
"reflect"
"strconv"
"strings"
"time"
"github.com/VividCortex/mysqlerr"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
"github.com/grafana/grafana/pkg/setting"
"github.com/go-sql-driver/mysql"
@@ -17,7 +19,12 @@ import (
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/tsdb/sqleng"
"xorm.io/core"
)
const (
dateFormat = "2006-01-02"
dateTimeFormat1 = "2006-01-02 15:04:05"
dateTimeFormat2 = "2006-01-02T15:04:05Z"
)
func characterEscape(s string, escapeChar string) string {
@@ -77,66 +84,6 @@ type mysqlQueryResultTransformer struct {
log log.Logger
}
func (t *mysqlQueryResultTransformer) TransformQueryResult(columnTypes []*sql.ColumnType, rows *core.Rows) (
plugins.DataRowValues, error) {
values := make([]interface{}, len(columnTypes))
for i := range values {
scanType := columnTypes[i].ScanType()
values[i] = reflect.New(scanType).Interface()
if columnTypes[i].DatabaseTypeName() == "BIT" {
values[i] = new([]byte)
}
}
if err := rows.Scan(values...); err != nil {
return nil, err
}
for i := 0; i < len(columnTypes); i++ {
typeName := reflect.ValueOf(values[i]).Type().String()
switch typeName {
case "*sql.RawBytes":
values[i] = string(*values[i].(*sql.RawBytes))
case "*mysql.NullTime":
sqlTime := (*values[i].(*mysql.NullTime))
if sqlTime.Valid {
values[i] = sqlTime.Time
} else {
values[i] = nil
}
case "*sql.NullInt64":
nullInt64 := (*values[i].(*sql.NullInt64))
if nullInt64.Valid {
values[i] = nullInt64.Int64
} else {
values[i] = nil
}
case "*sql.NullFloat64":
nullFloat64 := (*values[i].(*sql.NullFloat64))
if nullFloat64.Valid {
values[i] = nullFloat64.Float64
} else {
values[i] = nil
}
}
if columnTypes[i].DatabaseTypeName() == "DECIMAL" {
f, err := strconv.ParseFloat(values[i].(string), 64)
if err == nil {
values[i] = f
} else {
values[i] = nil
}
}
}
return values, nil
}
func (t *mysqlQueryResultTransformer) TransformQueryError(err error) error {
var driverErr *mysql.MySQLError
if errors.As(err, &driverErr) {
@@ -151,3 +98,199 @@ func (t *mysqlQueryResultTransformer) TransformQueryError(err error) error {
}
var errQueryFailed = errors.New("query failed - please inspect Grafana server log for details")
func (t *mysqlQueryResultTransformer) GetConverterList() []sqlutil.StringConverter {
// For the MySQL driver , we have these possible data types:
// https://www.w3schools.com/sql/sql_datatypes.asp#:~:text=In%20MySQL%20there%20are%20three,numeric%2C%20and%20date%20and%20time.
// Since by default, we convert all into String, we need only to handle the Numeric data types
return []sqlutil.StringConverter{
{
Name: "handle DOUBLE",
InputScanKind: reflect.Struct,
InputTypeName: "DOUBLE",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle BIGINT",
InputScanKind: reflect.Struct,
InputTypeName: "BIGINT",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableInt64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseInt(*in, 10, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle DECIMAL",
InputScanKind: reflect.Slice,
InputTypeName: "DECIMAL",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle DATETIME",
InputScanKind: reflect.Struct,
InputTypeName: "DATETIME",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableTime,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := time.Parse(dateTimeFormat1, *in)
if err == nil {
return &v, nil
}
v, err = time.Parse(dateTimeFormat2, *in)
if err == nil {
return &v, nil
}
return nil, err
},
},
},
{
Name: "handle DATE",
InputScanKind: reflect.Struct,
InputTypeName: "DATE",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableTime,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := time.Parse(dateFormat, *in)
if err == nil {
return &v, nil
}
v, err = time.Parse(dateTimeFormat1, *in)
if err == nil {
return &v, nil
}
v, err = time.Parse(dateTimeFormat2, *in)
if err == nil {
return &v, nil
}
return nil, err
},
},
},
{
Name: "handle TIMESTAMP",
InputScanKind: reflect.Struct,
InputTypeName: "TIMESTAMP",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableTime,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := time.Parse(dateTimeFormat1, *in)
if err == nil {
return &v, nil
}
v, err = time.Parse(dateTimeFormat2, *in)
if err == nil {
return &v, nil
}
return nil, err
},
},
},
{
Name: "handle YEAR",
InputScanKind: reflect.Struct,
InputTypeName: "YEAR",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableInt64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseInt(*in, 10, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle INT",
InputScanKind: reflect.Struct,
InputTypeName: "INT",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableInt64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseInt(*in, 10, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle FLOAT",
InputScanKind: reflect.Struct,
InputTypeName: "FLOAT",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
}
}
pkg/tsdb/mysql/mysql_test.go
+992 -940
View File
File diff suppressed because it is too large Load Diff
pkg/tsdb/postgres/postgres.go
+65 -40
View File
@@ -1,11 +1,13 @@
package postgres
import (
"database/sql"
"fmt"
"reflect"
"strconv"
"strings"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
"github.com/grafana/grafana/pkg/registry"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/util/errutil"
@@ -14,7 +16,6 @@ import (
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/tsdb/sqleng"
"xorm.io/core"
)
func init() {
@@ -114,7 +115,6 @@ func (s *PostgresService) generateConnectionString(datasource *models.DataSource
connStr += fmt.Sprintf(" sslmode='%s'", escape(tlsSettings.Mode))
// Attach root certificate if provided
// Attach root certificate if provided
if tlsSettings.RootCertFile != "" {
s.logger.Debug("Setting server root certificate", "tlsRootCert", tlsSettings.RootCertFile)
@@ -137,43 +137,68 @@ type postgresQueryResultTransformer struct {
log log.Logger
}
func (t *postgresQueryResultTransformer) TransformQueryResult(columnTypes []*sql.ColumnType, rows *core.Rows) (
plugins.DataRowValues, error) {
values := make([]interface{}, len(columnTypes))
valuePtrs := make([]interface{}, len(columnTypes))
for i := 0; i < len(columnTypes); i++ {
valuePtrs[i] = &values[i]
}
if err := rows.Scan(valuePtrs...); err != nil {
return nil, err
}
// convert types not handled by lib/pq
// unhandled types are returned as []byte
for i := 0; i < len(columnTypes); i++ {
if value, ok := values[i].([]byte); ok {
switch columnTypes[i].DatabaseTypeName() {
case "NUMERIC":
if v, err := strconv.ParseFloat(string(value), 64); err == nil {
values[i] = v
} else {
t.log.Debug("Rows", "Error converting numeric to float", value)
}
case "UNKNOWN", "CIDR", "INET", "MACADDR":
// char literals have type UNKNOWN
values[i] = string(value)
default:
t.log.Debug("Rows", "Unknown database type", columnTypes[i].DatabaseTypeName(), "value", value)
values[i] = string(value)
}
}
}
return values, nil
}
func (t *postgresQueryResultTransformer) TransformQueryError(err error) error {
return err
}
func (t *postgresQueryResultTransformer) GetConverterList() []sqlutil.StringConverter {
return []sqlutil.StringConverter{
{
Name: "handle FLOAT4",
InputScanKind: reflect.Interface,
InputTypeName: "FLOAT4",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle FLOAT8",
InputScanKind: reflect.Interface,
InputTypeName: "FLOAT8",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
{
Name: "handle NUMERIC",
InputScanKind: reflect.Interface,
InputTypeName: "NUMERIC",
ConversionFunc: func(in *string) (*string, error) { return in, nil },
Replacer: &sqlutil.StringFieldReplacer{
OutputFieldType: data.FieldTypeNullableFloat64,
ReplaceFunc: func(in *string) (interface{}, error) {
if in == nil {
return nil, nil
}
v, err := strconv.ParseFloat(*in, 64)
if err != nil {
return nil, err
}
return &v, nil
},
},
},
}
}
pkg/tsdb/postgres/postgres_test.go
+173 -123
View File
@@ -10,6 +10,7 @@ import (
"testing"
"time"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/components/securejsondata"
"github.com/grafana/grafana/pkg/components/simplejson"
"github.com/grafana/grafana/pkg/infra/log"
@@ -151,10 +152,9 @@ func TestGenerateConnectionString(t *testing.T) {
// devenv/README.md for setup instructions.
func TestPostgres(t *testing.T) {
// change to true to run the PostgreSQL tests
runPostgresTests := false
// runPostgresTests := true
const runPostgresTests = false
if !sqlstore.IsTestDbPostgres() && !runPostgresTests {
if !(sqlstore.IsTestDbPostgres() || runPostgresTests) {
t.Skip()
}
@@ -213,7 +213,7 @@ func TestPostgres(t *testing.T) {
c12_date date,
c13_time time without time zone,
c14_timetz time with time zone,
time date,
c15_interval interval
);
`
@@ -226,7 +226,7 @@ func TestPostgres(t *testing.T) {
4.5,6.7,1.1,1.2,
'char10','varchar10','text',
now(),now(),now(),now(),now(),'15m'::interval
now(),now(),now(),now(),now(),now(),'15m'::interval
);
`
_, err = sess.Exec(sql)
@@ -250,32 +250,36 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
column := queryResult.Tables[0].Rows[0]
require.Equal(t, int64(1), column[0].(int64))
require.Equal(t, int64(2), column[1].(int64))
require.Equal(t, int64(3), column[2].(int64))
frames, _ := queryResult.Dataframes.Decoded()
require.Len(t, frames, 1)
require.Len(t, frames[0].Fields, 17)
require.Equal(t, float64(4.5), column[3].(float64))
require.Equal(t, float64(6.7), column[4].(float64))
require.Equal(t, float64(1.1), column[5].(float64))
require.Equal(t, float64(1.2), column[6].(float64))
require.Equal(t, int16(1), *frames[0].Fields[0].At(0).(*int16))
require.Equal(t, int32(2), *frames[0].Fields[1].At(0).(*int32))
require.Equal(t, int64(3), *frames[0].Fields[2].At(0).(*int64))
require.Equal(t, "char10 ", column[7].(string))
require.Equal(t, "varchar10", column[8].(string))
require.Equal(t, "text", column[9].(string))
require.Equal(t, float64(4.5), *frames[0].Fields[3].At(0).(*float64))
require.Equal(t, float64(6.7), *frames[0].Fields[4].At(0).(*float64))
require.Equal(t, float64(1.1), *frames[0].Fields[5].At(0).(*float64))
require.Equal(t, float64(1.2), *frames[0].Fields[6].At(0).(*float64))
_, ok := column[10].(time.Time)
require.True(t, ok)
_, ok = column[11].(time.Time)
require.True(t, ok)
_, ok = column[12].(time.Time)
require.True(t, ok)
_, ok = column[13].(time.Time)
require.True(t, ok)
_, ok = column[14].(time.Time)
require.True(t, ok)
require.Equal(t, "char10 ", *frames[0].Fields[7].At(0).(*string))
require.Equal(t, "varchar10", *frames[0].Fields[8].At(0).(*string))
require.Equal(t, "text", *frames[0].Fields[9].At(0).(*string))
require.Equal(t, "00:15:00", column[15].(string))
_, ok := frames[0].Fields[10].At(0).(*time.Time)
require.True(t, ok)
_, ok = frames[0].Fields[11].At(0).(*time.Time)
require.True(t, ok)
_, ok = frames[0].Fields[12].At(0).(*time.Time)
require.True(t, ok)
_, ok = frames[0].Fields[13].At(0).(*time.Time)
require.True(t, ok)
_, ok = frames[0].Fields[14].At(0).(*time.Time)
require.True(t, ok)
_, ok = frames[0].Fields[15].At(0).(*time.Time)
require.True(t, ok)
require.Equal(t, "00:15:00", *frames[0].Fields[16].At(0).(*string))
})
})
@@ -335,26 +339,27 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
points := queryResult.Series[0].Points
frames, _ := queryResult.Dataframes.Decoded()
require.Len(t, frames, 1)
require.Equal(t, 4, frames[0].Fields[0].Len())
// without fill this should result in 4 buckets
require.Len(t, points, 4)
dt := fromStart
for i := 0; i < 2; i++ {
aValue := points[i][0].Float64
aTime := time.Unix(int64(points[i][1].Float64)/1000, 0)
aValue := *frames[0].Fields[1].At(i).(*float64)
aTime := *frames[0].Fields[0].At(i).(*time.Time)
require.Equal(t, float64(15), aValue)
require.Equal(t, dt, aTime)
require.Equal(t, int64(0), aTime.Unix()%300)
dt = dt.Add(5 * time.Minute)
}
// adjust for 10 minute gap between first and second set of points
dt = dt.Add(10 * time.Minute)
for i := 2; i < 4; i++ {
aValue := points[i][0].Float64
aTime := time.Unix(int64(points[i][1].Float64)/1000, 0)
aValue := *frames[0].Fields[1].At(i).(*float64)
aTime := *frames[0].Fields[0].At(i).(*time.Time)
require.Equal(t, float64(20), aValue)
require.Equal(t, dt, aTime)
dt = dt.Add(5 * time.Minute)
@@ -388,10 +393,12 @@ func TestPostgres(t *testing.T) {
resp, err := exe.DataQuery(context.Background(), nil, query)
require.NoError(t, err)
queryResult := resp.Results["A"]
frames, _ := queryResult.Dataframes.Decoded()
require.NoError(t, queryResult.Error)
require.Equal(t,
"SELECT floor(extract(epoch from time)/60)*60 AS time, avg(value) as value FROM metric GROUP BY 1 ORDER BY 1",
queryResult.Meta.Get(sqleng.MetaKeyExecutedQueryString).MustString())
frames[0].Meta.ExecutedQueryString)
})
t.Run("When doing a metric query using timeGroup with NULL fill enabled", func(t *testing.T) {
@@ -416,35 +423,36 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
points := queryResult.Series[0].Points
require.Len(t, points, 7)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 7, frames[0].Fields[0].Len())
dt := fromStart
for i := 0; i < 2; i++ {
aValue := points[i][0].Float64
aTime := time.Unix(int64(points[i][1].Float64)/1000, 0)
aValue := *frames[0].Fields[1].At(i).(*float64)
aTime := *frames[0].Fields[0].At(i).(*time.Time)
require.Equal(t, float64(15), aValue)
require.Equal(t, dt, aTime)
require.True(t, aTime.Equal(dt))
dt = dt.Add(5 * time.Minute)
}
// check for NULL values inserted by fill
require.False(t, points[2][0].Valid)
require.False(t, points[3][0].Valid)
require.Nil(t, frames[0].Fields[1].At(2))
require.Nil(t, frames[0].Fields[1].At(3))
// adjust for 10 minute gap between first and second set of points
dt = dt.Add(10 * time.Minute)
for i := 4; i < 6; i++ {
aValue := points[i][0].Float64
aTime := time.Unix(int64(points[i][1].Float64)/1000, 0)
aValue := *frames[0].Fields[1].At(i).(*float64)
aTime := *frames[0].Fields[0].At(i).(*time.Time)
require.Equal(t, float64(20), aValue)
require.Equal(t, dt, aTime)
require.True(t, aTime.Equal(dt))
dt = dt.Add(5 * time.Minute)
}
// check for NULL values inserted by fill
require.False(t, points[6][0].Valid)
require.Nil(t, frames[0].Fields[1].At(6))
})
t.Run("When doing a metric query using timeGroup with value fill enabled", func(t *testing.T) {
@@ -469,8 +477,9 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
points := queryResult.Series[0].Points
require.Equal(t, float64(1.5), points[3][0].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 1.5, *frames[0].Fields[1].At(3).(*float64))
})
})
@@ -496,10 +505,11 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
points := queryResult.Series[0].Points
require.Equal(t, float64(15.0), points[2][0].Float64)
require.Equal(t, float64(15.0), points[3][0].Float64)
require.Equal(t, float64(20.0), points[6][0].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, float64(15.0), *frames[0].Fields[1].At(2).(*float64))
require.Equal(t, float64(15.0), *frames[0].Fields[1].At(3).(*float64))
require.Equal(t, float64(20.0), *frames[0].Fields[1].At(6).(*float64))
})
t.Run("Given a table with metrics having multiple values and measurements", func(t *testing.T) {
@@ -570,7 +580,7 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
t.Run(
"When doing a metric query using epoch (int64) as time column and value column (int64) should return metric with time in milliseconds",
"When doing a metric query using epoch (int64) as time column and value column (int64) should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -589,11 +599,12 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Equal(t, 1, len(queryResult.Series))
require.Equal(t, float64(tInitial.UnixNano()/1e6), queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Len(t, frames, 1)
require.True(t, tInitial.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query using epoch (int64 nullable) as time column and value column (int64 nullable,) should return metric with time in milliseconds",
t.Run("When doing a metric query using epoch (int64 nullable) as time column and value column (int64 nullable,) should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -612,11 +623,12 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 1)
require.Equal(t, float64(tInitial.UnixNano()/1e6), queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Len(t, frames, 1)
require.True(t, tInitial.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query using epoch (float64) as time column and value column (float64), should return metric with time in milliseconds",
t.Run("When doing a metric query using epoch (float64) as time column and value column (float64), should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -635,11 +647,12 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 1)
require.Equal(t, float64(tInitial.UnixNano()/1e6), queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Len(t, frames, 1)
require.True(t, tInitial.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query using epoch (float64 nullable) as time column and value column (float64 nullable), should return metric with time in milliseconds",
t.Run("When doing a metric query using epoch (float64 nullable) as time column and value column (float64 nullable), should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -658,11 +671,12 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 1)
require.Equal(t, float64(tInitial.UnixNano()/1e6), queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.True(t, tInitial.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query using epoch (int32) as time column and value column (int32), should return metric with time in milliseconds",
t.Run("When doing a metric query using epoch (int32) as time column and value column (int32), should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -681,11 +695,12 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 1)
require.Equal(t, float64(tInitial.UnixNano()/1e6), queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.True(t, tInitial.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query using epoch (int32 nullable) as time column and value column (int32 nullable), should return metric with time in milliseconds",
t.Run("When doing a metric query using epoch (int32 nullable) as time column and value column (int32 nullable), should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -704,11 +719,12 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 1)
require.Equal(t, float64(tInitial.UnixNano()/1e6), queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.True(t, tInitial.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query using epoch (float32) as time column and value column (float32), should return metric with time in milliseconds",
t.Run("When doing a metric query using epoch (float32) as time column and value column (float32), should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -727,11 +743,13 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 1)
require.Equal(t, float64(float32(tInitial.Unix()))*1e3, queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
aTime := time.Unix(0, int64(float64(float32(tInitial.Unix()))*1e3)*int64(time.Millisecond))
require.True(t, aTime.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query using epoch (float32 nullable) as time column and value column (float32 nullable), should return metric with time in milliseconds",
t.Run("When doing a metric query using epoch (float32 nullable) as time column and value column (float32 nullable), should return metric with time in time.Time",
func(t *testing.T) {
query := plugins.DataQuery{
Queries: []plugins.DataSubQuery{
@@ -750,8 +768,10 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 1)
require.Equal(t, float64(float32(tInitial.Unix()))*1e3, queryResult.Series[0].Points[0][1].Float64)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
aTime := time.Unix(0, int64(float64(float32(tInitial.Unix()))*1e3)*int64(time.Millisecond))
require.True(t, aTime.Equal(*frames[0].Fields[0].At(0).(*time.Time)))
})
t.Run("When doing a metric query grouping by time and select metric column should return correct series", func(t *testing.T) {
@@ -772,9 +792,11 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 2)
require.Equal(t, "Metric A - value one", queryResult.Series[0].Name)
require.Equal(t, "Metric B - value one", queryResult.Series[1].Name)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
require.Equal(t, data.Labels{"metric": "Metric A - value one"}, frames[0].Fields[1].Labels)
require.Equal(t, data.Labels{"metric": "Metric B - value one"}, frames[0].Fields[2].Labels)
})
t.Run("When doing a metric query with metric column and multiple value columns", func(t *testing.T) {
@@ -795,11 +817,18 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 4)
require.Equal(t, "Metric A valueOne", queryResult.Series[0].Name)
require.Equal(t, "Metric A valueTwo", queryResult.Series[1].Name)
require.Equal(t, "Metric B valueOne", queryResult.Series[2].Name)
require.Equal(t, "Metric B valueTwo", queryResult.Series[3].Name)
frames, err := queryResult.Dataframes.Decoded()
require.NoError(t, err)
require.Equal(t, 1, len(frames))
require.Equal(t, 5, len(frames[0].Fields))
require.Equal(t, "valueOne", frames[0].Fields[1].Name)
require.Equal(t, data.Labels{"metric": "Metric A"}, frames[0].Fields[1].Labels)
require.Equal(t, "valueOne", frames[0].Fields[2].Name)
require.Equal(t, data.Labels{"metric": "Metric B"}, frames[0].Fields[2].Labels)
require.Equal(t, "valueTwo", frames[0].Fields[3].Name)
require.Equal(t, data.Labels{"metric": "Metric A"}, frames[0].Fields[3].Labels)
require.Equal(t, "valueTwo", frames[0].Fields[4].Name)
require.Equal(t, data.Labels{"metric": "Metric B"}, frames[0].Fields[4].Labels)
})
t.Run("When doing a metric query grouping by time should return correct series", func(t *testing.T) {
@@ -820,9 +849,11 @@ func TestPostgres(t *testing.T) {
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Series, 2)
require.Equal(t, "valueOne", queryResult.Series[0].Name)
require.Equal(t, "valueTwo", queryResult.Series[1].Name)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
require.Equal(t, "valueOne", frames[0].Fields[1].Name)
require.Equal(t, "valueTwo", frames[0].Fields[2].Name)
})
t.Run("When doing a query with timeFrom,timeTo,unixEpochFrom,unixEpochTo macros", func(t *testing.T) {
@@ -850,9 +881,11 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
frames, _ := queryResult.Dataframes.Decoded()
require.Len(t, frames, 1)
require.Equal(t,
"SELECT time FROM metric_values WHERE time > '2018-03-15T12:55:00Z' OR time < '2018-03-15T12:55:00Z' OR 1 < 1521118500 OR 1521118800 > 1 ORDER BY 1",
queryResult.Meta.Get(sqleng.MetaKeyExecutedQueryString).MustString())
frames[0].Meta.ExecutedQueryString)
})
})
@@ -910,7 +943,10 @@ func TestPostgres(t *testing.T) {
resp, err := exe.DataQuery(context.Background(), nil, query)
queryResult := resp.Results["Deploys"]
require.NoError(t, err)
require.Len(t, queryResult.Tables[0].Rows, 3)
frames, _ := queryResult.Dataframes.Decoded()
require.Len(t, frames, 1)
require.Len(t, frames[0].Fields, 3)
})
t.Run("When doing an annotation query of ticket events should return expected result", func(t *testing.T) {
@@ -933,7 +969,10 @@ func TestPostgres(t *testing.T) {
resp, err := exe.DataQuery(context.Background(), nil, query)
queryResult := resp.Results["Tickets"]
require.NoError(t, err)
require.Len(t, queryResult.Tables[0].Rows, 3)
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
})
t.Run("When doing an annotation query with a time column in datetime format", func(t *testing.T) {
@@ -960,14 +999,15 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Tables[0].Rows, 1)
columns := queryResult.Tables[0].Rows[0]
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
//Should be in milliseconds
require.Equal(t, float64(dt.UnixNano()/1e6), columns[0].(float64))
// Should be in time.Time
require.Equal(t, dt.Unix(), (*frames[0].Fields[0].At(0).(*time.Time)).Unix())
})
t.Run("When doing an annotation query with a time column in epoch second format should return ms", func(t *testing.T) {
t.Run("When doing an annotation query with a time column in epoch second format should return time.Time", func(t *testing.T) {
dt := time.Date(2018, 3, 14, 21, 20, 6, 527e6, time.UTC)
query := plugins.DataQuery{
@@ -975,7 +1015,7 @@ func TestPostgres(t *testing.T) {
{
Model: simplejson.NewFromAny(map[string]interface{}{
"rawSql": fmt.Sprintf(`SELECT
%d as time,
%d as time,
'message' as text,
'tag1,tag2' as tags
`, dt.Unix()),
@@ -990,14 +1030,16 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Tables[0].Rows, 1)
columns := queryResult.Tables[0].Rows[0]
//Should be in milliseconds
require.Equal(t, dt.Unix()*1000, columns[0].(int64))
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
// Should be in time.Time
require.Equal(t, dt.Unix(), (*frames[0].Fields[0].At(0).(*time.Time)).Unix())
})
t.Run("When doing an annotation query with a time column in epoch second format (t *testing.Tint) should return ms", func(t *testing.T) {
t.Run("When doing an annotation query with a time column in epoch second format (t *testing.Tint) should return time.Time", func(t *testing.T) {
dt := time.Date(2018, 3, 14, 21, 20, 6, 527e6, time.UTC)
query := plugins.DataQuery{
@@ -1005,7 +1047,7 @@ func TestPostgres(t *testing.T) {
{
Model: simplejson.NewFromAny(map[string]interface{}{
"rawSql": fmt.Sprintf(`SELECT
cast(%d as bigint) as time,
cast(%d as bigint) as time,
'message' as text,
'tag1,tag2' as tags
`, dt.Unix()),
@@ -1020,14 +1062,16 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Tables[0].Rows, 1)
columns := queryResult.Tables[0].Rows[0]
//Should be in milliseconds
require.Equal(t, dt.Unix()*1000, columns[0].(int64))
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
// Should be in time.Time
require.Equal(t, dt.Unix(), (*frames[0].Fields[0].At(0).(*time.Time)).Unix())
})
t.Run("When doing an annotation query with a time column in epoch millisecond format should return ms", func(t *testing.T) {
t.Run("When doing an annotation query with a time column in epoch millisecond format should return time.Time", func(t *testing.T) {
dt := time.Date(2018, 3, 14, 21, 20, 6, 527e6, time.UTC)
query := plugins.DataQuery{
@@ -1050,11 +1094,13 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Tables[0].Rows, 1)
columns := queryResult.Tables[0].Rows[0]
//Should be in milliseconds
require.Equal(t, dt.Unix()*1000, columns[0].(int64))
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
// Should be in time.Time
require.Equal(t, dt.Unix(), (*frames[0].Fields[0].At(0).(*time.Time)).Unix())
})
t.Run("When doing an annotation query with a time column holding a bigint null value should return nil", func(t *testing.T) {
@@ -1078,11 +1124,13 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Tables[0].Rows, 1)
columns := queryResult.Tables[0].Rows[0]
//Should be in milliseconds
require.Nil(t, columns[0])
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
// Should be in time.Time
require.Nil(t, frames[0].Fields[0].At(0))
})
t.Run("When doing an annotation query with a time column holding a timestamp null value should return nil", func(t *testing.T) {
@@ -1091,10 +1139,10 @@ func TestPostgres(t *testing.T) {
{
Model: simplejson.NewFromAny(map[string]interface{}{
"rawSql": `SELECT
cast(null as timestamp) as time,
'message' as text,
'tag1,tag2' as tags
`,
cast(null as timestamp) as time,
'message' as text,
'tag1,tag2' as tags
`,
"format": "table",
}),
RefID: "A",
@@ -1106,11 +1154,13 @@ func TestPostgres(t *testing.T) {
require.NoError(t, err)
queryResult := resp.Results["A"]
require.NoError(t, queryResult.Error)
require.Len(t, queryResult.Tables[0].Rows, 1)
columns := queryResult.Tables[0].Rows[0]
//Should be in milliseconds
assert.Nil(t, columns[0])
frames, _ := queryResult.Dataframes.Decoded()
require.Equal(t, 1, len(frames))
require.Equal(t, 3, len(frames[0].Fields))
// Should be in time.Time
assert.Nil(t, frames[0].Fields[0].At(0))
})
})
}
pkg/tsdb/sqleng/resample.go
+136
View File
@@ -0,0 +1,136 @@
package sqleng
import (
"fmt"
"time"
"github.com/grafana/grafana-plugin-sdk-go/data"
)
// getRowFillValues populates a slice of values corresponding to the provided data.Frame fields.
// Uses data.FillMissing settings to fill in values that are missing. Values are normally missing
// due to that the selected query interval doesn't match the intervals of the data returned from
// the query and therefore needs to be resampled.
func getRowFillValues(f *data.Frame, tsSchema data.TimeSeriesSchema, currentTime time.Time,
fillMissing *data.FillMissing, intermediateRows []int, lastSeenRowIdx int) []interface{} {
vals := make([]interface{}, 0, len(f.Fields))
for i, field := range f.Fields {
// if the current field is the time index of the series
// set the new value to be added to the new timestamp
if i == tsSchema.TimeIndex {
switch f.Fields[tsSchema.TimeIndex].Type() {
case data.FieldTypeTime:
vals = append(vals, currentTime)
default:
vals = append(vals, &currentTime)
}
continue
}
isValueField := false
for _, idx := range tsSchema.ValueIndices {
if i == idx {
isValueField = true
break
}
}
// if the current field is value Field
// set the new value to the last seen field value (if such exists)
// otherwise set the appropriate value according to the fillMissing mode
// if the current field is string field)
// set the new value to be added to the last seen value (if such exists)
// if the Frame is wide then there should not be any string fields
var newVal interface{}
if isValueField {
if len(intermediateRows) > 0 {
// instead of setting the last seen
// we could set avg, sum, min or max
// of the intermediate values for each field
newVal = f.At(i, intermediateRows[len(intermediateRows)-1])
} else {
val, err := data.GetMissing(fillMissing, field, lastSeenRowIdx)
if err == nil {
newVal = val
}
}
} else if lastSeenRowIdx >= 0 {
newVal = f.At(i, lastSeenRowIdx)
}
vals = append(vals, newVal)
}
return vals
}
// resample resample provided time-series data.Frame.
// This is needed in the case of the selected query interval doesn't
// match the intervals of the time-series field in the data.Frame and
// therefore needs to be resampled.
func resample(f *data.Frame, qm dataQueryModel) (*data.Frame, error) {
tsSchema := f.TimeSeriesSchema()
if tsSchema.Type == data.TimeSeriesTypeNot {
return f, fmt.Errorf("can not fill missing, not timeseries frame")
}
if qm.Interval == 0 {
return f, nil
}
newFields := make([]*data.Field, 0, len(f.Fields))
for _, field := range f.Fields {
newField := data.NewFieldFromFieldType(field.Type(), 0)
newField.Name = field.Name
newField.Labels = field.Labels
newFields = append(newFields, newField)
}
resampledFrame := data.NewFrame(f.Name, newFields...)
resampledFrame.Meta = f.Meta
resampledRowidx := 0
lastSeenRowIdx := -1
timeField := f.Fields[tsSchema.TimeIndex]
for currentTime := qm.TimeRange.From; !currentTime.After(qm.TimeRange.To); currentTime = currentTime.Add(qm.Interval) {
initialRowIdx := 0
if lastSeenRowIdx > 0 {
initialRowIdx = lastSeenRowIdx + 1
}
intermediateRows := make([]int, 0)
for {
rowLen, err := f.RowLen()
if err != nil {
return f, err
}
if initialRowIdx == rowLen {
break
}
t, ok := timeField.ConcreteAt(initialRowIdx)
if !ok {
return f, fmt.Errorf("time point is nil")
}
if t.(time.Time).After(currentTime) {
nextTime := currentTime.Add(qm.Interval)
if t.(time.Time).Before(nextTime) {
intermediateRows = append(intermediateRows, initialRowIdx)
lastSeenRowIdx = initialRowIdx
initialRowIdx++
}
break
}
intermediateRows = append(intermediateRows, initialRowIdx)
lastSeenRowIdx = initialRowIdx
initialRowIdx++
}
// no intermediate points; set values following fill missing mode
fieldVals := getRowFillValues(f, tsSchema, currentTime, qm.FillMissing, intermediateRows, lastSeenRowIdx)
resampledFrame.InsertRow(resampledRowidx, fieldVals...)
resampledRowidx++
}
return resampledFrame, nil
}
pkg/tsdb/sqleng/resample_test.go
+309
View File
@@ -0,0 +1,309 @@
package sqleng
import (
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/stretchr/testify/require"
"github.com/xorcare/pointer"
)
func TestResampleWide(t *testing.T) {
tests := []struct {
name string
input *data.Frame
fillMissing *data.FillMissing
timeRange backend.TimeRange
interval time.Duration
output *data.Frame
}{
{
name: "interval 1s; fill null",
fillMissing: &data.FillMissing{Mode: data.FillModeNull},
timeRange: backend.TimeRange{
From: time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
To: time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
},
interval: time.Second,
input: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(10),
pointer.Int64(12),
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(10.5),
pointer.Float64(12.5),
pointer.Float64(15.0),
})),
output: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 25, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
nil,
pointer.Int64(10),
pointer.Int64(12),
nil,
nil,
nil,
pointer.Int64(15),
nil,
nil,
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
nil,
pointer.Float64(10.5),
pointer.Float64(12.5),
nil,
nil,
nil,
pointer.Float64(15.0),
nil,
nil,
})),
},
{
name: "interval 1s; fill value",
fillMissing: &data.FillMissing{Mode: data.FillModeValue, Value: -1},
timeRange: backend.TimeRange{
From: time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
To: time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
},
interval: time.Second,
input: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(10),
pointer.Int64(12),
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(10.5),
pointer.Float64(12.5),
pointer.Float64(15.0),
})),
output: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 25, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(-1),
pointer.Int64(10),
pointer.Int64(12),
pointer.Int64(-1),
pointer.Int64(-1),
pointer.Int64(-1),
pointer.Int64(15),
pointer.Int64(-1),
pointer.Int64(-1),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(-1),
pointer.Float64(10.5),
pointer.Float64(12.5),
pointer.Float64(-1),
pointer.Float64(-1),
pointer.Float64(-1),
pointer.Float64(15.0),
pointer.Float64(-1),
pointer.Float64(-1),
})),
},
{
name: "interval 1s; fill previous",
fillMissing: &data.FillMissing{Mode: data.FillModePrevious},
timeRange: backend.TimeRange{
From: time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
To: time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
},
interval: time.Second,
input: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(10),
pointer.Int64(12),
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(10.5),
pointer.Float64(12.5),
pointer.Float64(15.0),
})),
output: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 25, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
nil,
pointer.Int64(10),
pointer.Int64(12),
pointer.Int64(12),
pointer.Int64(12),
pointer.Int64(12),
pointer.Int64(15),
pointer.Int64(15),
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
nil,
pointer.Float64(10.5),
pointer.Float64(12.5),
pointer.Float64(12.5),
pointer.Float64(12.5),
pointer.Float64(12.5),
pointer.Float64(15.0),
pointer.Float64(15.0),
pointer.Float64(15.0),
})),
},
{
name: "interval 2s; fill null",
fillMissing: &data.FillMissing{Mode: data.FillModeNull},
timeRange: backend.TimeRange{
From: time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
To: time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
},
interval: 2 * time.Second,
input: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(10),
pointer.Int64(12),
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(10.5),
pointer.Float64(12.5),
pointer.Float64(15.0),
})),
output: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(12),
nil,
nil,
pointer.Int64(15),
nil,
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(12.5),
nil,
nil,
pointer.Float64(15.0),
nil,
})),
},
{
name: "interval 1s; fill null; rows outside timerange window",
fillMissing: &data.FillMissing{Mode: data.FillModeNull},
timeRange: backend.TimeRange{
From: time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
To: time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
},
interval: time.Second,
input: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(10),
pointer.Int64(12),
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(10.5),
pointer.Float64(12.5),
pointer.Float64(15.0),
})),
output: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(12),
nil,
nil,
nil,
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(12.5),
nil,
nil,
nil,
pointer.Float64(15.0),
})),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
frame, err := resample(tt.input, dataQueryModel{
FillMissing: tt.fillMissing,
TimeRange: tt.timeRange,
Interval: tt.interval,
})
require.NoError(t, err)
if diff := cmp.Diff(tt.output, frame, data.FrameTestCompareOptions()...); diff != "" {
t.Errorf("Result mismatch (-want +got):\n%s", diff)
}
})
}
}
pkg/tsdb/sqleng/sql_engine.go
+666 -523
View File
File diff suppressed because it is too large Load Diff
pkg/tsdb/sqleng/sql_engine_test.go
+265 -184
View File
@@ -7,19 +7,20 @@ import (
"testing"
"time"
"github.com/grafana/grafana/pkg/components/null"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
"github.com/grafana/grafana/pkg/components/simplejson"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/plugins"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/xorcare/pointer"
"xorm.io/core"
)
func TestSQLEngine(t *testing.T) {
dt := time.Date(2018, 3, 14, 21, 20, 6, int(527345*time.Microsecond), time.UTC)
earlyDt := time.Date(1970, 3, 14, 21, 20, 6, int(527345*time.Microsecond), time.UTC)
t.Run("Given a time range between 2018-04-12 00:00 and 2018-04-12 00:05", func(t *testing.T) {
from := time.Date(2018, 4, 12, 18, 0, 0, 0, time.UTC)
@@ -58,56 +59,48 @@ func TestSQLEngine(t *testing.T) {
})
})
t.Run("Given row values with time.Time as time columns", func(t *testing.T) {
var nilPointer *time.Time
fixtures := make([]interface{}, 5)
fixtures[0] = dt
fixtures[1] = &dt
fixtures[2] = earlyDt
fixtures[3] = &earlyDt
fixtures[4] = nilPointer
for i := range fixtures {
ConvertSqlTimeColumnToEpochMs(fixtures, i)
}
expected := float64(dt.UnixNano()) / float64(time.Millisecond)
expectedEarly := float64(earlyDt.UnixNano()) / float64(time.Millisecond)
require.Equal(t, expected, fixtures[0].(float64))
require.Equal(t, expected, fixtures[1].(float64))
require.Equal(t, expectedEarly, fixtures[2].(float64))
require.Equal(t, expectedEarly, fixtures[3].(float64))
require.Nil(t, fixtures[4])
})
t.Run("Given row values with int64 as time columns", func(t *testing.T) {
tSeconds := dt.Unix()
tMilliseconds := dt.UnixNano() / 1e6
tNanoSeconds := dt.UnixNano()
var nilPointer *int64
fixtures := make([]interface{}, 7)
fixtures[0] = tSeconds
fixtures[1] = &tSeconds
fixtures[2] = tMilliseconds
fixtures[3] = &tMilliseconds
fixtures[4] = tNanoSeconds
fixtures[5] = &tNanoSeconds
fixtures[6] = nilPointer
originFrame := data.NewFrame("",
data.NewField("time1", nil, []int64{
tSeconds,
}),
data.NewField("time2", nil, []*int64{
pointer.Int64(tSeconds),
}),
data.NewField("time3", nil, []int64{
tMilliseconds,
}),
data.NewField("time4", nil, []*int64{
pointer.Int64(tMilliseconds),
}),
data.NewField("time5", nil, []int64{
tNanoSeconds,
}),
data.NewField("time6", nil, []*int64{
pointer.Int64(tNanoSeconds),
}),
data.NewField("time7", nil, []*int64{
nilPointer,
}),
)
for i := range fixtures {
ConvertSqlTimeColumnToEpochMs(fixtures, i)
for i := 0; i < len(originFrame.Fields); i++ {
err := convertSQLTimeColumnToEpochMS(originFrame, i)
require.NoError(t, err)
}
require.Equal(t, tSeconds*1e3, fixtures[0].(int64))
require.Equal(t, tSeconds*1e3, fixtures[1].(int64))
require.Equal(t, tMilliseconds, fixtures[2].(int64))
require.Equal(t, tMilliseconds, fixtures[3].(int64))
require.Equal(t, tMilliseconds, fixtures[4].(int64))
require.Equal(t, tMilliseconds, fixtures[5].(int64))
require.Nil(t, fixtures[6])
require.Equal(t, dt.Unix(), (*originFrame.Fields[0].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[1].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[2].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[3].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[4].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[5].At(0).(*time.Time)).Unix())
require.Nil(t, originFrame.Fields[6].At(0))
})
t.Run("Given row values with uint64 as time columns", func(t *testing.T) {
@@ -116,62 +109,91 @@ func TestSQLEngine(t *testing.T) {
tNanoSeconds := uint64(dt.UnixNano())
var nilPointer *uint64
fixtures := make([]interface{}, 7)
fixtures[0] = tSeconds
fixtures[1] = &tSeconds
fixtures[2] = tMilliseconds
fixtures[3] = &tMilliseconds
fixtures[4] = tNanoSeconds
fixtures[5] = &tNanoSeconds
fixtures[6] = nilPointer
originFrame := data.NewFrame("",
data.NewField("time1", nil, []uint64{
tSeconds,
}),
data.NewField("time2", nil, []*uint64{
pointer.Uint64(tSeconds),
}),
data.NewField("time3", nil, []uint64{
tMilliseconds,
}),
data.NewField("time4", nil, []*uint64{
pointer.Uint64(tMilliseconds),
}),
data.NewField("time5", nil, []uint64{
tNanoSeconds,
}),
data.NewField("time6", nil, []*uint64{
pointer.Uint64(tNanoSeconds),
}),
data.NewField("time7", nil, []*uint64{
nilPointer,
}),
)
for i := range fixtures {
ConvertSqlTimeColumnToEpochMs(fixtures, i)
for i := 0; i < len(originFrame.Fields); i++ {
err := convertSQLTimeColumnToEpochMS(originFrame, i)
require.NoError(t, err)
}
require.Equal(t, int64(tSeconds*1e3), fixtures[0].(int64))
require.Equal(t, int64(tSeconds*1e3), fixtures[1].(int64))
require.Equal(t, int64(tMilliseconds), fixtures[2].(int64))
require.Equal(t, int64(tMilliseconds), fixtures[3].(int64))
require.Equal(t, int64(tMilliseconds), fixtures[4].(int64))
require.Equal(t, int64(tMilliseconds), fixtures[5].(int64))
require.Nil(t, fixtures[6])
require.Equal(t, dt.Unix(), (*originFrame.Fields[0].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[1].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[2].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[3].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[4].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[5].At(0).(*time.Time)).Unix())
require.Nil(t, originFrame.Fields[6].At(0))
})
t.Run("Given row values with int32 as time columns", func(t *testing.T) {
tSeconds := int32(dt.Unix())
var nilInt *int32
fixtures := make([]interface{}, 3)
fixtures[0] = tSeconds
fixtures[1] = &tSeconds
fixtures[2] = nilInt
for i := range fixtures {
ConvertSqlTimeColumnToEpochMs(fixtures, i)
originFrame := data.NewFrame("",
data.NewField("time1", nil, []int32{
tSeconds,
}),
data.NewField("time2", nil, []*int32{
pointer.Int32(tSeconds),
}),
data.NewField("time7", nil, []*int32{
nilInt,
}),
)
for i := 0; i < 3; i++ {
err := convertSQLTimeColumnToEpochMS(originFrame, i)
require.NoError(t, err)
}
require.Equal(t, dt.Unix()*1e3, fixtures[0].(int64))
require.Equal(t, dt.Unix()*1e3, fixtures[1].(int64))
require.Nil(t, fixtures[2])
require.Equal(t, dt.Unix(), (*originFrame.Fields[0].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[1].At(0).(*time.Time)).Unix())
require.Nil(t, originFrame.Fields[2].At(0))
})
t.Run("Given row values with uint32 as time columns", func(t *testing.T) {
tSeconds := uint32(dt.Unix())
var nilInt *uint32
fixtures := make([]interface{}, 3)
fixtures[0] = tSeconds
fixtures[1] = &tSeconds
fixtures[2] = nilInt
for i := range fixtures {
ConvertSqlTimeColumnToEpochMs(fixtures, i)
originFrame := data.NewFrame("",
data.NewField("time1", nil, []uint32{
tSeconds,
}),
data.NewField("time2", nil, []*uint32{
pointer.Uint32(tSeconds),
}),
data.NewField("time7", nil, []*uint32{
nilInt,
}),
)
for i := 0; i < len(originFrame.Fields); i++ {
err := convertSQLTimeColumnToEpochMS(originFrame, i)
require.NoError(t, err)
}
require.Equal(t, dt.Unix()*1e3, fixtures[0].(int64))
require.Equal(t, dt.Unix()*1e3, fixtures[1].(int64))
require.Nil(t, fixtures[2])
require.Equal(t, dt.Unix(), (*originFrame.Fields[0].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[1].At(0).(*time.Time)).Unix())
require.Nil(t, originFrame.Fields[2].At(0))
})
t.Run("Given row values with float64 as time columns", func(t *testing.T) {
@@ -180,137 +202,192 @@ func TestSQLEngine(t *testing.T) {
tNanoSeconds := float64(dt.UnixNano())
var nilPointer *float64
fixtures := make([]interface{}, 7)
fixtures[0] = tSeconds
fixtures[1] = &tSeconds
fixtures[2] = tMilliseconds
fixtures[3] = &tMilliseconds
fixtures[4] = tNanoSeconds
fixtures[5] = &tNanoSeconds
fixtures[6] = nilPointer
originFrame := data.NewFrame("",
data.NewField("time1", nil, []float64{
tSeconds,
}),
data.NewField("time2", nil, []*float64{
pointer.Float64(tSeconds),
}),
data.NewField("time3", nil, []float64{
tMilliseconds,
}),
data.NewField("time4", nil, []*float64{
pointer.Float64(tMilliseconds),
}),
data.NewField("time5", nil, []float64{
tNanoSeconds,
}),
data.NewField("time6", nil, []*float64{
pointer.Float64(tNanoSeconds),
}),
data.NewField("time7", nil, []*float64{
nilPointer,
}),
)
for i := range fixtures {
ConvertSqlTimeColumnToEpochMs(fixtures, i)
for i := 0; i < len(originFrame.Fields); i++ {
err := convertSQLTimeColumnToEpochMS(originFrame, i)
require.NoError(t, err)
}
require.Equal(t, tMilliseconds, fixtures[0].(float64))
require.Equal(t, tMilliseconds, fixtures[1].(float64))
require.Equal(t, tMilliseconds, fixtures[2].(float64))
require.Equal(t, tMilliseconds, fixtures[3].(float64))
require.Equal(t, tMilliseconds, fixtures[4].(float64))
require.Equal(t, tMilliseconds, fixtures[5].(float64))
require.Nil(t, fixtures[6])
require.Equal(t, dt.Unix(), (*originFrame.Fields[0].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[1].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[2].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[3].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[4].At(0).(*time.Time)).Unix())
require.Equal(t, dt.Unix(), (*originFrame.Fields[5].At(0).(*time.Time)).Unix())
require.Nil(t, originFrame.Fields[6].At(0))
})
t.Run("Given row values with float32 as time columns", func(t *testing.T) {
tSeconds := float32(dt.Unix())
var nilInt *float32
fixtures := make([]interface{}, 3)
fixtures[0] = tSeconds
fixtures[1] = &tSeconds
fixtures[2] = nilInt
for i := range fixtures {
ConvertSqlTimeColumnToEpochMs(fixtures, i)
originFrame := data.NewFrame("",
data.NewField("time1", nil, []float32{
tSeconds,
}),
data.NewField("time2", nil, []*float32{
pointer.Float32(tSeconds),
}),
data.NewField("time7", nil, []*float32{
nilInt,
}),
)
for i := 0; i < len(originFrame.Fields); i++ {
err := convertSQLTimeColumnToEpochMS(originFrame, i)
require.NoError(t, err)
}
require.Equal(t, float64(tSeconds)*1e3, fixtures[0].(float64))
require.Equal(t, float64(tSeconds)*1e3, fixtures[1].(float64))
require.Nil(t, fixtures[2])
require.Equal(t, int64(tSeconds), (*originFrame.Fields[0].At(0).(*time.Time)).Unix())
require.Equal(t, int64(tSeconds), (*originFrame.Fields[1].At(0).(*time.Time)).Unix())
require.Nil(t, originFrame.Fields[2].At(0))
})
t.Run("Given row with value columns", func(t *testing.T) {
intValue := 1
int64Value := int64(1)
int32Value := int32(1)
int16Value := int16(1)
int8Value := int8(1)
float64Value := float64(1)
float32Value := float32(1)
uintValue := uint(1)
uint64Value := uint64(1)
uint32Value := uint32(1)
uint16Value := uint16(1)
uint8Value := uint8(1)
testCases := []struct {
name string
value interface{}
}{
{"intValue", intValue},
{"&intValue", &intValue},
{"int64Value", int64Value},
{"&int64Value", &int64Value},
{"int32Value", int32Value},
{"&int32Value", &int32Value},
{"int16Value", int16Value},
{"&int16Value", &int16Value},
{"int8Value", int8Value},
{"&int8Value", &int8Value},
{"float64Value", float64Value},
{"&float64Value", &float64Value},
{"float32Value", float32Value},
{"&float32Value", &float32Value},
{"uintValue", uintValue},
{"&uintValue", &uintValue},
{"uint64Value", uint64Value},
{"&uint64Value", &uint64Value},
{"uint32Value", uint32Value},
{"&uint32Value", &uint32Value},
{"uint16Value", uint16Value},
{"&uint16Value", &uint16Value},
{"uint8Value", uint8Value},
{"&uint8Value", &uint8Value},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
value, err := ConvertSqlValueColumnToFloat("col", tc.value)
require.NoError(t, err)
require.True(t, value.Valid)
require.Equal(t, null.FloatFrom(1).Float64, value.Float64)
})
t.Run("Given row with value columns, would be converted to float64", func(t *testing.T) {
originFrame := data.NewFrame("",
data.NewField("value1", nil, []int64{
int64(1),
}),
data.NewField("value2", nil, []*int64{
pointer.Int64(1),
}),
data.NewField("value3", nil, []int32{
int32(1),
}),
data.NewField("value4", nil, []*int32{
pointer.Int32(1),
}),
data.NewField("value5", nil, []int16{
int16(1),
}),
data.NewField("value6", nil, []*int16{
pointer.Int16(1),
}),
data.NewField("value7", nil, []int8{
int8(1),
}),
data.NewField("value8", nil, []*int8{
pointer.Int8(1),
}),
data.NewField("value9", nil, []float64{
float64(1),
}),
data.NewField("value10", nil, []*float64{
pointer.Float64(1),
}),
data.NewField("value11", nil, []float32{
float32(1),
}),
data.NewField("value12", nil, []*float32{
pointer.Float32(1),
}),
data.NewField("value13", nil, []uint64{
uint64(1),
}),
data.NewField("value14", nil, []*uint64{
pointer.Uint64(1),
}),
data.NewField("value15", nil, []uint32{
uint32(1),
}),
data.NewField("value16", nil, []*uint32{
pointer.Uint32(1),
}),
data.NewField("value17", nil, []uint16{
uint16(1),
}),
data.NewField("value18", nil, []*uint16{
pointer.Uint16(1),
}),
data.NewField("value19", nil, []uint8{
uint8(1),
}),
data.NewField("value20", nil, []*uint8{
pointer.Uint8(1),
}),
)
for i := 0; i < len(originFrame.Fields); i++ {
_, err := convertSQLValueColumnToFloat(originFrame, i)
require.NoError(t, err)
if i == 8 {
require.Equal(t, float64(1), originFrame.Fields[i].At(0).(float64))
} else {
require.NotNil(t, originFrame.Fields[i].At(0).(*float64))
require.Equal(t, float64(1), *originFrame.Fields[i].At(0).(*float64))
}
}
})
t.Run("Given row with nil value columns", func(t *testing.T) {
var intNilPointer *int
var int64NilPointer *int64
var int32NilPointer *int32
var int16NilPointer *int16
var int8NilPointer *int8
var float64NilPointer *float64
var float32NilPointer *float32
var uintNilPointer *uint
var uint64NilPointer *uint64
var uint32NilPointer *uint32
var uint16NilPointer *uint16
var uint8NilPointer *uint8
testCases := []struct {
name string
value interface{}
}{
{"intNilPointer", intNilPointer},
{"int64NilPointer", int64NilPointer},
{"int32NilPointer", int32NilPointer},
{"int16NilPointer", int16NilPointer},
{"int8NilPointer", int8NilPointer},
{"float64NilPointer", float64NilPointer},
{"float32NilPointer", float32NilPointer},
{"uintNilPointer", uintNilPointer},
{"uint64NilPointer", uint64NilPointer},
{"uint32NilPointer", uint32NilPointer},
{"uint16NilPointer", uint16NilPointer},
{"uint8NilPointer", uint8NilPointer},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
value, err := ConvertSqlValueColumnToFloat("col", tc.value)
originFrame := data.NewFrame("",
data.NewField("value1", nil, []*int64{
int64NilPointer,
}),
data.NewField("value2", nil, []*int32{
int32NilPointer,
}),
data.NewField("value3", nil, []*int16{
int16NilPointer,
}),
data.NewField("value4", nil, []*int8{
int8NilPointer,
}),
data.NewField("value5", nil, []*float64{
float64NilPointer,
}),
data.NewField("value6", nil, []*float32{
float32NilPointer,
}),
data.NewField("value7", nil, []*uint64{
uint64NilPointer,
}),
data.NewField("value8", nil, []*uint32{
uint32NilPointer,
}),
data.NewField("value9", nil, []*uint16{
uint16NilPointer,
}),
data.NewField("value10", nil, []*uint8{
uint8NilPointer,
}),
)
for i := 0; i < len(originFrame.Fields); i++ {
t.Run("", func(t *testing.T) {
_, err := convertSQLValueColumnToFloat(originFrame, i)
require.NoError(t, err)
require.False(t, value.Valid)
require.Nil(t, originFrame.Fields[i].At(0))
})
}
})
@@ -352,3 +429,7 @@ func (t *testQueryResultTransformer) TransformQueryError(err error) error {
t.transformQueryErrorWasCalled = true
return err
}
func (t *testQueryResultTransformer) GetConverterList() []sqlutil.StringConverter {
return nil
}
pkg/tsdb/sqleng/trim.go
+51
View File
@@ -0,0 +1,51 @@
package sqleng
import (
"fmt"
"time"
"github.com/grafana/grafana-plugin-sdk-go/data"
)
// trim trims rows that are outside the qm.TimeRange.
func trim(f *data.Frame, qm dataQueryModel) error {
tsSchema := f.TimeSeriesSchema()
if tsSchema.Type == data.TimeSeriesTypeNot {
return fmt.Errorf("can not trim non-timeseries frame")
}
timeField := f.Fields[tsSchema.TimeIndex]
if timeField.Len() == 0 {
return nil
}
// Trim rows after end
for i := timeField.Len() - 1; i >= 0; i-- {
t, ok := timeField.ConcreteAt(i)
if !ok {
return fmt.Errorf("time point is nil")
}
if !t.(time.Time).After(qm.TimeRange.To) {
break
}
f.DeleteRow(i)
}
// Trim rows before start
for timeField.Len() > 0 {
t, ok := timeField.ConcreteAt(0)
if !ok {
return fmt.Errorf("time point is nil")
}
if !t.(time.Time).Before(qm.TimeRange.From) {
break
}
f.DeleteRow(0)
}
return nil
}
pkg/tsdb/sqleng/trim_test.go
+171
View File
@@ -0,0 +1,171 @@
package sqleng
import (
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/stretchr/testify/require"
"github.com/xorcare/pointer"
)
func TestTrimWide(t *testing.T) {
tests := []struct {
name string
input *data.Frame
timeRange backend.TimeRange
output *data.Frame
}{
{
name: "needs trimming",
timeRange: backend.TimeRange{
From: time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
To: time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
},
input: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 25, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
nil,
pointer.Int64(10),
pointer.Int64(12),
nil,
nil,
nil,
pointer.Int64(15),
nil,
nil,
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
nil,
pointer.Float64(10.5),
pointer.Float64(12.5),
nil,
nil,
nil,
pointer.Float64(15.0),
nil,
nil,
})),
output: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
pointer.Int64(12),
nil,
nil,
nil,
pointer.Int64(15),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
pointer.Float64(12.5),
nil,
nil,
nil,
pointer.Float64(15.0),
})),
},
{
name: "does not need trimming",
timeRange: backend.TimeRange{
From: time.Date(2020, 1, 2, 3, 4, 15, 0, time.UTC),
To: time.Date(2020, 1, 2, 3, 4, 30, 0, time.UTC),
},
input: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 25, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
nil,
pointer.Int64(10),
pointer.Int64(12),
nil,
nil,
nil,
pointer.Int64(15),
nil,
nil,
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
nil,
pointer.Float64(10.5),
pointer.Float64(12.5),
nil,
nil,
nil,
pointer.Float64(15.0),
nil,
nil,
})),
output: data.NewFrame("wide_test",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 18, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 19, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 20, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 21, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 22, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 23, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 24, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 25, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 26, 0, time.UTC),
}),
data.NewField("Values Ints", nil, []*int64{
nil,
pointer.Int64(10),
pointer.Int64(12),
nil,
nil,
nil,
pointer.Int64(15),
nil,
nil,
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []*float64{
nil,
pointer.Float64(10.5),
pointer.Float64(12.5),
nil,
nil,
nil,
pointer.Float64(15.0),
nil,
nil,
})),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := trim(tt.input, dataQueryModel{
TimeRange: tt.timeRange,
})
require.NoError(t, err)
if diff := cmp.Diff(tt.output, tt.input, data.FrameTestCompareOptions()...); diff != "" {
t.Errorf("Result mismatch (-want +got):\n%s", diff)
}
})
}
}