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

fix: Fixes data race in execute test

pull/10616/head
Nathaniel Cook 2018-08-27 13:57:26 -06:00
parent a2e7ba3422
commit 575c8138cc
1 changed files with 28 additions and 17 deletions
query/execute

45
query/execute/executor_test.go
View File

@ -44,8 +44,8 @@ func TestExecutor_Execute(t *testing.T) {
Procedures: map[plan.ProcedureID]*plan.Procedure{ Procedures: map[plan.ProcedureID]*plan.Procedure{
plan.ProcedureIDFromOperationID("from"): { plan.ProcedureIDFromOperationID("from"): {
ID: plan.ProcedureIDFromOperationID("from"), ID: plan.ProcedureIDFromOperationID("from"),
Spec: &testFromProcedureSource{ Spec: newTestFromProcedureSource(
data: []query.Table{&executetest.Table{ []*executetest.Table{&executetest.Table{
KeyCols: []string{"_start", "_stop"}, KeyCols: []string{"_start", "_stop"},
ColMeta: []query.ColMeta{ ColMeta: []query.ColMeta{
{Label: "_start", Type: query.TTime}, {Label: "_start", Type: query.TTime},
@ -61,7 +61,7 @@ func TestExecutor_Execute(t *testing.T) {
{execute.Time(0), execute.Time(5), execute.Time(4), 5.0}, {execute.Time(0), execute.Time(5), execute.Time(4), 5.0},
}, },
}}, }},
}, ),
Bounds: &plan.BoundsSpec{ Bounds: &plan.BoundsSpec{
Start: values.ConvertTime(time.Unix(0, 1)), Start: values.ConvertTime(time.Unix(0, 1)),
Stop: values.ConvertTime(time.Unix(0, 5)), Stop: values.ConvertTime(time.Unix(0, 5)),
@ -114,8 +114,8 @@ func TestExecutor_Execute(t *testing.T) {
Procedures: map[plan.ProcedureID]*plan.Procedure{ Procedures: map[plan.ProcedureID]*plan.Procedure{
plan.ProcedureIDFromOperationID("from"): { plan.ProcedureIDFromOperationID("from"): {
ID: plan.ProcedureIDFromOperationID("from"), ID: plan.ProcedureIDFromOperationID("from"),
Spec: &testFromProcedureSource{ Spec: newTestFromProcedureSource(
data: []query.Table{&executetest.Table{ []*executetest.Table{&executetest.Table{
KeyCols: []string{"_start", "_stop"}, KeyCols: []string{"_start", "_stop"},
ColMeta: []query.ColMeta{ ColMeta: []query.ColMeta{
{Label: "_start", Type: query.TTime}, {Label: "_start", Type: query.TTime},
@ -131,7 +131,7 @@ func TestExecutor_Execute(t *testing.T) {
{execute.Time(0), execute.Time(5), execute.Time(4), int64(5)}, {execute.Time(0), execute.Time(5), execute.Time(4), int64(5)},
}, },
}}, }},
}, ),
Bounds: &plan.BoundsSpec{ Bounds: &plan.BoundsSpec{
Start: values.ConvertTime(time.Unix(0, 1)), Start: values.ConvertTime(time.Unix(0, 1)),
Stop: values.ConvertTime(time.Unix(0, 5)), Stop: values.ConvertTime(time.Unix(0, 5)),
@ -218,8 +218,8 @@ func TestExecutor_Execute(t *testing.T) {
Procedures: map[plan.ProcedureID]*plan.Procedure{ Procedures: map[plan.ProcedureID]*plan.Procedure{
plan.ProcedureIDFromOperationID("from"): { plan.ProcedureIDFromOperationID("from"): {
ID: plan.ProcedureIDFromOperationID("from"), ID: plan.ProcedureIDFromOperationID("from"),
Spec: &testFromProcedureSource{ Spec: newTestFromProcedureSource(
data: []query.Table{ []*executetest.Table{
&executetest.Table{ &executetest.Table{
KeyCols: []string{"_start", "_stop", "_key"}, KeyCols: []string{"_start", "_stop", "_key"},
ColMeta: []query.ColMeta{ ColMeta: []query.ColMeta{
@ -286,7 +286,7 @@ func TestExecutor_Execute(t *testing.T) {
}, },
}, },
}, },
}, ),
Parents: nil, Parents: nil,
Bounds: &plan.BoundsSpec{ Bounds: &plan.BoundsSpec{
Start: values.ConvertTime(time.Unix(0, 1)), Start: values.ConvertTime(time.Unix(0, 1)),
@ -437,8 +437,8 @@ func TestExecutor_Execute(t *testing.T) {
Procedures: map[plan.ProcedureID]*plan.Procedure{ Procedures: map[plan.ProcedureID]*plan.Procedure{
plan.ProcedureIDFromOperationID("from"): { plan.ProcedureIDFromOperationID("from"): {
ID: plan.ProcedureIDFromOperationID("from"), ID: plan.ProcedureIDFromOperationID("from"),
Spec: &testFromProcedureSource{ Spec: newTestFromProcedureSource(
data: []query.Table{&executetest.Table{ []*executetest.Table{&executetest.Table{
KeyCols: []string{"_start", "_stop"}, KeyCols: []string{"_start", "_stop"},
ColMeta: []query.ColMeta{ ColMeta: []query.ColMeta{
{Label: "_start", Type: query.TTime}, {Label: "_start", Type: query.TTime},
@ -454,7 +454,7 @@ func TestExecutor_Execute(t *testing.T) {
{execute.Time(0), execute.Time(5), execute.Time(4), 5.0}, {execute.Time(0), execute.Time(5), execute.Time(4), 5.0},
}, },
}}, }},
}, ),
Bounds: &plan.BoundsSpec{ Bounds: &plan.BoundsSpec{
Start: values.ConvertTime(time.Unix(0, 1)), Start: values.ConvertTime(time.Unix(0, 1)),
Stop: values.ConvertTime(time.Unix(0, 5)), Stop: values.ConvertTime(time.Unix(0, 5)),
@ -565,10 +565,21 @@ func TestExecutor_Execute(t *testing.T) {
} }
type testFromProcedureSource struct { type testFromProcedureSource struct {
data []query.Table data []*executetest.Table
ts []execute.Transformation ts []execute.Transformation
} }
func newTestFromProcedureSource(data []*executetest.Table) *testFromProcedureSource {
p := &testFromProcedureSource{
data: data,
}
// Normalize the data before anything can read it
for _, tbl := range p.data {
tbl.Normalize()
}
return p
}
func (p *testFromProcedureSource) Kind() plan.ProcedureKind { func (p *testFromProcedureSource) Kind() plan.ProcedureKind {
return "from-test" return "from-test"
} }
@ -585,11 +596,11 @@ func (p *testFromProcedureSource) Run(ctx context.Context) {
id := execute.DatasetID(uuid.NewV4()) id := execute.DatasetID(uuid.NewV4())
for _, t := range p.ts { for _, t := range p.ts {
var max execute.Time var max execute.Time
for _, b := range p.data { for _, tbl := range p.data {
t.Process(id, b) t.Process(id, tbl)
stopIdx := execute.ColIdx(execute.DefaultStopColLabel, b.Cols()) stopIdx := execute.ColIdx(execute.DefaultStopColLabel, tbl.Cols())
if stopIdx >= 0 { if stopIdx >= 0 {
if s := b.Key().ValueTime(stopIdx); s > max { if s := tbl.Key().ValueTime(stopIdx); s > max {
max = s max = s
} }
} }