Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
influxdb/query/subquery_test.go

370 lines
15 KiB
Go
Raw Normal View History

Add some unit tests to subqueries This is not complete, but it is a starting point for more thorough tests of subqueries. This also reorders the use of `cmp.Diff` so the `want` is first and `got` is second. This way, the `want` shows up as a minus sign in the diff rather than, confusingly, as a plus sign.
2018-03-27 19:56:27 +00:00
package query_test
import (
"context"
"testing"
"github.com/google/go-cmp/cmp"
Pass the query authorizer to subqueries The query authorizer was not being properly passed to subqueries so rejections did not happen when a subquery was the one reading the value. Similarly, the max series limit was not being propagated downwards either.
2018-12-07 21:08:24 +00:00
"github.com/influxdata/influxdb/models"
Add some unit tests to subqueries This is not complete, but it is a starting point for more thorough tests of subqueries. This also reorders the use of `cmp.Diff` so the `want` is first and `got` is second. This way, the `want` shows up as a minus sign in the diff rather than, confusingly, as a plus sign.
2018-03-27 19:56:27 +00:00
"github.com/influxdata/influxdb/query"
"github.com/influxdata/influxql"
)
type CreateIteratorFn func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator
func TestSubquery(t *testing.T) {
for _, test := range []struct {
Name string
Statement string
Fields map[string]influxql.DataType
MapShardsFn func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn
Rows []query.Row
}{
{
Name: "AuxiliaryFields",
Statement: `SELECT max / 2.0 FROM (SELECT max(value) FROM cpu GROUP BY time(5s)) WHERE time >= '1970-01-01T00:00:00Z' AND time < '1970-01-01T00:00:15Z'`,
Fields: map[string]influxql.DataType{"value": influxql.Float},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
if got, want := tr.MinTimeNano(), 0*Second; got != want {
t.Errorf("unexpected min time: got=%d want=%d", got, want)
}
if got, want := tr.MaxTimeNano(), 15*Second-1; got != want {
t.Errorf("unexpected max time: got=%d want=%d", got, want)
}
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
if got, want := m.Name, "cpu"; got != want {
t.Errorf("unexpected source: got=%s want=%s", got, want)
}
if got, want := opt.Expr.String(), "max(value::float)"; got != want {
t.Errorf("unexpected expression: got=%s want=%s", got, want)
}
return &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Time: 0 * Second, Value: 5},
{Name: "cpu", Time: 5 * Second, Value: 3},
{Name: "cpu", Time: 10 * Second, Value: 8},
}}
}
},
Rows: []query.Row{
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{2.5}},
{Time: 5 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{1.5}},
{Time: 10 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{float64(4)}},
},
},
{
Name: "AuxiliaryFields_WithWhereClause",
Statement: `SELECT host FROM (SELECT max(value), host FROM cpu GROUP BY time(5s)) WHERE max > 4 AND time >= '1970-01-01T00:00:00Z' AND time < '1970-01-01T00:00:15Z'`,
Fields: map[string]influxql.DataType{
"value": influxql.Float,
"host": influxql.Tag,
},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
if got, want := tr.MinTimeNano(), 0*Second; got != want {
t.Errorf("unexpected min time: got=%d want=%d", got, want)
}
if got, want := tr.MaxTimeNano(), 15*Second-1; got != want {
t.Errorf("unexpected max time: got=%d want=%d", got, want)
}
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
if got, want := m.Name, "cpu"; got != want {
t.Errorf("unexpected source: got=%s want=%s", got, want)
}
if got, want := opt.Expr.String(), "max(value::float)"; got != want {
t.Errorf("unexpected expression: got=%s want=%s", got, want)
}
if got, want := opt.Aux, []influxql.VarRef{{Val: "host", Type: influxql.Tag}}; !cmp.Equal(got, want) {
t.Errorf("unexpected auxiliary fields:\n%s", cmp.Diff(want, got))
}
return &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Time: 0 * Second, Value: 5, Aux: []interface{}{"server02"}},
{Name: "cpu", Time: 5 * Second, Value: 3, Aux: []interface{}{"server01"}},
{Name: "cpu", Time: 10 * Second, Value: 8, Aux: []interface{}{"server03"}},
}}
}
},
Rows: []query.Row{
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{"server02"}},
{Time: 10 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{"server03"}},
},
},
{
Name: "AuxiliaryFields_NonExistentField",
Statement: `SELECT host FROM (SELECT max(value) FROM cpu GROUP BY time(5s)) WHERE time >= '1970-01-01T00:00:00Z' AND time < '1970-01-01T00:00:15Z'`,
Fields: map[string]influxql.DataType{"value": influxql.Float},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
return &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Time: 0 * Second, Value: 5},
{Name: "cpu", Time: 5 * Second, Value: 3},
{Name: "cpu", Time: 10 * Second, Value: 8},
}}
}
},
Rows: []query.Row(nil),
},
{
Name: "AggregateOfMath",
Statement: `SELECT mean(percentage) FROM (SELECT value * 100.0 AS percentage FROM cpu) WHERE time >= '1970-01-01T00:00:00Z' AND time < '1970-01-01T00:00:15Z' GROUP BY time(5s)`,
Fields: map[string]influxql.DataType{"value": influxql.Float},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
if got, want := tr.MinTimeNano(), 0*Second; got != want {
t.Errorf("unexpected min time: got=%d want=%d", got, want)
}
if got, want := tr.MaxTimeNano(), 15*Second-1; got != want {
t.Errorf("unexpected max time: got=%d want=%d", got, want)
}
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
if got, want := m.Name, "cpu"; got != want {
t.Errorf("unexpected source: got=%s want=%s", got, want)
}
if got, want := opt.Expr, influxql.Expr(nil); got != want {
t.Errorf("unexpected expression: got=%s want=%s", got, want)
}
if got, want := opt.Aux, []influxql.VarRef{{Val: "value", Type: influxql.Float}}; !cmp.Equal(got, want) {
t.Errorf("unexpected auxiliary fields:\n%s", cmp.Diff(want, got))
}
return &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Time: 0 * Second, Aux: []interface{}{0.5}},
{Name: "cpu", Time: 2 * Second, Aux: []interface{}{1.0}},
{Name: "cpu", Time: 5 * Second, Aux: []interface{}{0.05}},
{Name: "cpu", Time: 8 * Second, Aux: []interface{}{0.45}},
{Name: "cpu", Time: 12 * Second, Aux: []interface{}{0.34}},
}}
}
},
Rows: []query.Row{
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{float64(75)}},
{Time: 5 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{float64(25)}},
{Time: 10 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{float64(34)}},
},
},
Enable casting values from a subquery This also fixes the cursor system to abandon iterators that will not produce meaningful results since the variables are all unknown types. This creates a weird behavior that existed in previous releases and we are keeping here for backwards compatibility. If a subquery referenced a field that didn't exist in the subquery, it will return nothing. But, if there are two subqueries and one of them has the field exist and the other doesn't, the second will return all null values.
2018-03-30 21:58:37 +00:00
{
Name: "Cast",
Statement: `SELECT value::integer FROM (SELECT mean(value) AS value FROM cpu)`,
Fields: map[string]influxql.DataType{"value": influxql.Integer},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
if got, want := m.Name, "cpu"; got != want {
t.Errorf("unexpected source: got=%s want=%s", got, want)
}
if got, want := opt.Expr.String(), "mean(value::integer)"; got != want {
t.Errorf("unexpected expression: got=%s want=%s", got, want)
}
return &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Time: 0 * Second, Value: float64(20) / float64(6)},
}}
}
},
Fix subquery functionality when a function references a tag from the subquery It has previously been allowed for a subquery to use a tag within a function (such as `count()`) when the tag is from a subquery and the subquery itself references a field at some point to perform the join. This functionality regressed in 1.6 because of a change in how subqueries were executed that forgot to treat a tag the same as a string field. This fixes that regression and adds a test case to avoid hitting that regression again.
2018-10-04 14:57:15 +00:00
Rows: []query.Row{
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{int64(3)}},
},
},
{
Name: "CountTag",
Statement: `SELECT count(host) FROM (SELECT value, host FROM cpu) WHERE time >= '1970-01-01T00:00:00Z' AND time < '1970-01-01T00:00:15Z'`,
Fields: map[string]influxql.DataType{
"value": influxql.Float,
"host": influxql.Tag,
},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
if got, want := tr.MinTimeNano(), 0*Second; got != want {
t.Errorf("unexpected min time: got=%d want=%d", got, want)
}
if got, want := tr.MaxTimeNano(), 15*Second-1; got != want {
t.Errorf("unexpected max time: got=%d want=%d", got, want)
}
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
if got, want := m.Name, "cpu"; got != want {
t.Errorf("unexpected source: got=%s want=%s", got, want)
}
if got, want := opt.Aux, []influxql.VarRef{
{Val: "host", Type: influxql.Tag},
{Val: "value", Type: influxql.Float},
}; !cmp.Equal(got, want) {
t.Errorf("unexpected auxiliary fields:\n%s", cmp.Diff(want, got))
}
return &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Aux: []interface{}{"server01", 5.0}},
{Name: "cpu", Aux: []interface{}{"server02", 3.0}},
{Name: "cpu", Aux: []interface{}{"server03", 8.0}},
}}
}
},
Enable casting values from a subquery This also fixes the cursor system to abandon iterators that will not produce meaningful results since the variables are all unknown types. This creates a weird behavior that existed in previous releases and we are keeping here for backwards compatibility. If a subquery referenced a field that didn't exist in the subquery, it will return nothing. But, if there are two subqueries and one of them has the field exist and the other doesn't, the second will return all null values.
2018-03-30 21:58:37 +00:00
Rows: []query.Row{
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{int64(3)}},
},
},
Strip tags from a subquery when the outer query does not group by that tag The following would, erroneously, not strip the tag from the inner query: SELECT value FROM (SELECT value FROM cpu GROUP BY host) The inner query was supposed to group by the host tag, but the outer query should strip it away since it is not being grouped by anymore. This fixes things so that the result will have the tags stripped away when they are not requested in the grouping.
2018-10-04 14:59:41 +00:00
{
Name: "StripTags",
Statement: `SELECT max FROM (SELECT max(value) FROM cpu GROUP BY host) WHERE time >= '1970-01-01T00:00:00Z' AND time < '1970-01-01T00:00:15Z'`,
Fields: map[string]influxql.DataType{"value": influxql.Float},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
if got, want := tr.MinTimeNano(), 0*Second; got != want {
t.Errorf("unexpected min time: got=%d want=%d", got, want)
}
if got, want := tr.MaxTimeNano(), 15*Second-1; got != want {
t.Errorf("unexpected max time: got=%d want=%d", got, want)
}
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
if got, want := m.Name, "cpu"; got != want {
t.Errorf("unexpected source: got=%s want=%s", got, want)
}
if got, want := opt.Expr.String(), "max(value::float)"; got != want {
t.Errorf("unexpected expression: got=%s want=%s", got, want)
}
return &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Tags: ParseTags("host=server01"), Value: 5},
{Name: "cpu", Tags: ParseTags("host=server02"), Value: 3},
{Name: "cpu", Tags: ParseTags("host=server03"), Value: 8},
}}
}
},
Rows: []query.Row{
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{5.0}},
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{3.0}},
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{8.0}},
},
},
Fix the sort order for aggregates so that they are sorted by tag and then time The reduce iterators would read in the points for a window, which matched the grouping of the outermost query, and then it would sort them by the time before emitting the points. When there were multiple series, this would sometimes cause a conflict because it would change the sorting of the inner query output when selectors were used within a subquery. Then, these emitted points would be output in the wrong order and they wouldn't join correctly when multiple cursors were used. This fixes it so the sorting happens per series grouping rather than on all of the points together so they retain their tag order which is the correct sorting method.
2019-04-16 15:01:30 +00:00
{
Name: "DifferentDimensionsWithSelectors",
Statement: `SELECT sum("max_min") FROM (
SELECT max("value") - min("value") FROM cpu GROUP BY time(30s), host
) WHERE time >= '1970-01-01T00:00:00Z' AND time < '1970-01-01T00:01:00Z' GROUP BY time(30s)`,
Fields: map[string]influxql.DataType{"value": influxql.Float},
MapShardsFn: func(t *testing.T, tr influxql.TimeRange) CreateIteratorFn {
if got, want := tr.MinTimeNano(), 0*Second; got != want {
t.Errorf("unexpected min time: got=%d want=%d", got, want)
}
if got, want := tr.MaxTimeNano(), 60*Second-1; got != want {
t.Errorf("unexpected max time: got=%d want=%d", got, want)
}
return func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) query.Iterator {
if got, want := m.Name, "cpu"; got != want {
t.Errorf("unexpected source: got=%s want=%s", got, want)
}
var itr query.Iterator = &FloatIterator{Points: []query.FloatPoint{
{Name: "cpu", Tags: ParseTags("host=A"), Time: 0 * Second, Value: 2},
{Name: "cpu", Tags: ParseTags("host=A"), Time: 10 * Second, Value: 7},
{Name: "cpu", Tags: ParseTags("host=A"), Time: 20 * Second, Value: 3},
{Name: "cpu", Tags: ParseTags("host=B"), Time: 0 * Second, Value: 8},
{Name: "cpu", Tags: ParseTags("host=B"), Time: 10 * Second, Value: 3},
{Name: "cpu", Tags: ParseTags("host=B"), Time: 20 * Second, Value: 7},
{Name: "cpu", Tags: ParseTags("host=A"), Time: 30 * Second, Value: 2},
{Name: "cpu", Tags: ParseTags("host=A"), Time: 40 * Second, Value: 1},
{Name: "cpu", Tags: ParseTags("host=A"), Time: 50 * Second, Value: 9},
{Name: "cpu", Tags: ParseTags("host=B"), Time: 30 * Second, Value: 2},
{Name: "cpu", Tags: ParseTags("host=B"), Time: 40 * Second, Value: 2},
{Name: "cpu", Tags: ParseTags("host=B"), Time: 50 * Second, Value: 2},
}}
if _, ok := opt.Expr.(*influxql.Call); ok {
i, err := query.NewCallIterator(itr, opt)
if err != nil {
panic(err)
}
itr = i
}
return itr
}
},
Rows: []query.Row{
{Time: 0 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{float64(10)}},
{Time: 30 * Second, Series: query.Series{Name: "cpu"}, Values: []interface{}{float64(8)}},
},
},
Add some unit tests to subqueries This is not complete, but it is a starting point for more thorough tests of subqueries. This also reorders the use of `cmp.Diff` so the `want` is first and `got` is second. This way, the `want` shows up as a minus sign in the diff rather than, confusingly, as a plus sign.
2018-03-27 19:56:27 +00:00
} {
t.Run(test.Name, func(t *testing.T) {
shardMapper := ShardMapper{
MapShardsFn: func(sources influxql.Sources, tr influxql.TimeRange) query.ShardGroup {
fn := test.MapShardsFn(t, tr)
return &ShardGroup{
Fields: test.Fields,
CreateIteratorFn: func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) (query.Iterator, error) {
return fn(ctx, m, opt), nil
},
}
},
}
stmt := MustParseSelectStatement(test.Statement)
stmt.OmitTime = true
cur, err := query.Select(context.Background(), stmt, &shardMapper, query.SelectOptions{})
if err != nil {
t.Fatalf("unexpected parse error: %s", err)
} else if a, err := ReadCursor(cur); err != nil {
t.Fatalf("unexpected error: %s", err)
} else if diff := cmp.Diff(test.Rows, a); diff != "" {
t.Fatalf("unexpected points:\n%s", diff)
}
})
}
}
Pass the query authorizer to subqueries The query authorizer was not being properly passed to subqueries so rejections did not happen when a subquery was the one reading the value. Similarly, the max series limit was not being propagated downwards either.
2018-12-07 21:08:24 +00:00
type openAuthorizer struct{}
func (*openAuthorizer) AuthorizeDatabase(p influxql.Privilege, name string) bool { return true }
func (*openAuthorizer) AuthorizeQuery(database string, query *influxql.Query) error { return nil }
func (*openAuthorizer) AuthorizeSeriesRead(database string, measurement []byte, tags models.Tags) bool {
return true
}
func (*openAuthorizer) AuthorizeSeriesWrite(database string, measurement []byte, tags models.Tags) bool {
return true
}
// Ensure that the subquery gets passed the query authorizer.
func TestSubquery_Authorizer(t *testing.T) {
auth := &openAuthorizer{}
shardMapper := ShardMapper{
MapShardsFn: func(sources influxql.Sources, tr influxql.TimeRange) query.ShardGroup {
return &ShardGroup{
Fields: map[string]influxql.DataType{
"value": influxql.Float,
},
CreateIteratorFn: func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) (query.Iterator, error) {
if opt.Authorizer != auth {
t.Errorf("query authorizer has not been set")
}
return nil, nil
},
}
},
}
stmt := MustParseSelectStatement(`SELECT max(value) FROM (SELECT value FROM cpu)`)
cur, err := query.Select(context.Background(), stmt, &shardMapper, query.SelectOptions{
Authorizer: auth,
})
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
cur.Close()
}
// Ensure that the subquery gets passed the max series limit.
func TestSubquery_MaxSeriesN(t *testing.T) {
shardMapper := ShardMapper{
MapShardsFn: func(sources influxql.Sources, tr influxql.TimeRange) query.ShardGroup {
return &ShardGroup{
Fields: map[string]influxql.DataType{
"value": influxql.Float,
},
CreateIteratorFn: func(ctx context.Context, m *influxql.Measurement, opt query.IteratorOptions) (query.Iterator, error) {
if opt.MaxSeriesN != 1000 {
t.Errorf("max series limit has not been set")
}
return nil, nil
},
}
},
}
stmt := MustParseSelectStatement(`SELECT max(value) FROM (SELECT value FROM cpu)`)
cur, err := query.Select(context.Background(), stmt, &shardMapper, query.SelectOptions{
MaxSeriesN: 1000,
})
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
cur.Close()
}