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

Merge pull request #9551 from influxdata/er-fieldset-panic 

Fix panic when checking fieldsets
pull/9557/merge
Edd Robinson 2018-03-12 17:28:58 +00:00 committed by GitHub
parent ad3e5becfa bdd61298ea
commit 7c3ae91d1e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 74 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
services/subscriber/service_test.go
View File

@ -185,6 +185,7 @@ func TestService_ModeALL(t *testing.T) {
}
func TestService_ModeANY(t *testing.T) {
t.Skip("TODO: flaky test.")
dataChanged := make(chan struct{})
ms := MetaClient{}
ms.WaitForDataChangedFn = func() chan struct{} {

2
tsdb/engine/tsm1/engine.go
View File

@ -2643,7 +2643,7 @@ func (e *Engine) buildCursor(ctx context.Context, measurement, seriesKey string,
}
// Look up fields for measurement.
mf := e.fieldset.Fields(measurement)
mf := e.fieldset.FieldsByString(measurement)
if mf == nil {
return nil
}

2
tsdb/engine/tsm1/engine_cursor.go
View File

@ -12,7 +12,7 @@ import (
func (e *Engine) CreateCursor(ctx context.Context, r *tsdb.CursorRequest) (tsdb.Cursor, error) {
// Look up fields for measurement.
mf := e.fieldset.Fields(r.Measurement)
mf := e.fieldset.FieldsByString(r.Measurement)
if mf == nil {
return nil, nil
}

62
tsdb/index.go
View File

@ -1092,8 +1092,9 @@ func (itr *tagValueMergeIterator) Next() (_ []byte, err error) {
// IndexSet represents a list of indexes.
type IndexSet struct {
Indexes []Index
SeriesFile *SeriesFile
Indexes []Index // The set of indexes comprising this IndexSet.
SeriesFile *SeriesFile // The Series File associated with the db for this set.
fieldSets []*MeasurementFieldSet // field sets for _all_ indexes in this set's DB.
}
// Database returns the database name of the first index.
@ -1104,20 +1105,40 @@ func (is IndexSet) Database() string {
return is.Indexes[0].Database()
}
// FieldSet returns the fieldset of the first index.
func (is IndexSet) FieldSet() *MeasurementFieldSet {
// HasField determines if any of the field sets on the set of indexes in the
// IndexSet have the provided field for the provided measurement.
func (is IndexSet) HasField(measurement []byte, field string) bool {
if len(is.Indexes) == 0 {
return nil
return false
}
return is.Indexes[0].FieldSet()
if len(is.fieldSets) == 0 {
// field sets may not have been initialised yet.
is.fieldSets = make([]*MeasurementFieldSet, 0, len(is.Indexes))
for _, idx := range is.Indexes {
is.fieldSets = append(is.fieldSets, idx.FieldSet())
}
}
for _, fs := range is.fieldSets {
if fs.Fields(measurement).HasField(field) {
return true
}
}
return false
}
// DedupeInmemIndexes returns an index set which removes duplicate in-memory indexes.
func (is IndexSet) DedupeInmemIndexes() IndexSet {
other := IndexSet{Indexes: make([]Index, 0, len(is.Indexes)), SeriesFile: is.SeriesFile}
other := IndexSet{
Indexes: make([]Index, 0, len(is.Indexes)),
SeriesFile: is.SeriesFile,
fieldSets: make([]*MeasurementFieldSet, 0, len(is.Indexes)),
}
var hasInmem bool
for _, idx := range is.Indexes {
other.fieldSets = append(other.fieldSets, idx.FieldSet())
if idx.Type() == "inmem" {
if !hasInmem {
other.Indexes = append(other.Indexes, idx)
@ -1699,9 +1720,8 @@ func (is IndexSet) measurementSeriesByExprIterator(name []byte, expr influxql.Ex
if expr == nil {
return is.measurementSeriesIDIterator(name)
}
fieldset := is.FieldSet()
itr, err := is.seriesByExprIterator(name, expr, fieldset.CreateFieldsIfNotExists(name))
itr, err := is.seriesByExprIterator(name, expr)
if err != nil {
return nil, err
}
@ -1754,19 +1774,19 @@ func (is IndexSet) MeasurementSeriesKeysByExpr(name []byte, expr influxql.Expr)
return keys, nil
}
func (is IndexSet) seriesByExprIterator(name []byte, expr influxql.Expr, mf *MeasurementFields) (SeriesIDIterator, error) {
func (is IndexSet) seriesByExprIterator(name []byte, expr influxql.Expr) (SeriesIDIterator, error) {
switch expr := expr.(type) {
case *influxql.BinaryExpr:
switch expr.Op {
case influxql.AND, influxql.OR:
// Get the series IDs and filter expressions for the LHS.
litr, err := is.seriesByExprIterator(name, expr.LHS, mf)
litr, err := is.seriesByExprIterator(name, expr.LHS)
if err != nil {
return nil, err
}
// Get the series IDs and filter expressions for the RHS.
ritr, err := is.seriesByExprIterator(name, expr.RHS, mf)
ritr, err := is.seriesByExprIterator(name, expr.RHS)
if err != nil {
if litr != nil {
litr.Close()
@ -1783,11 +1803,11 @@ func (is IndexSet) seriesByExprIterator(name []byte, expr influxql.Expr, mf *Mea
return UnionSeriesIDIterators(litr, ritr), nil
default:
return is.seriesByBinaryExprIterator(name, expr, mf)
return is.seriesByBinaryExprIterator(name, expr)
}
case *influxql.ParenExpr:
return is.seriesByExprIterator(name, expr.Expr, mf)
return is.seriesByExprIterator(name, expr.Expr)
case *influxql.BooleanLiteral:
if expr.Val {
@ -1801,7 +1821,7 @@ func (is IndexSet) seriesByExprIterator(name []byte, expr influxql.Expr, mf *Mea
}
// seriesByBinaryExprIterator returns a series iterator and a filtering expression.
func (is IndexSet) seriesByBinaryExprIterator(name []byte, n *influxql.BinaryExpr, mf *MeasurementFields) (SeriesIDIterator, error) {
func (is IndexSet) seriesByBinaryExprIterator(name []byte, n *influxql.BinaryExpr) (SeriesIDIterator, error) {
// If this binary expression has another binary expression, then this
// is some expression math and we should just pass it to the underlying query.
if _, ok := n.LHS.(*influxql.BinaryExpr); ok {
@ -1830,7 +1850,7 @@ func (is IndexSet) seriesByBinaryExprIterator(name []byte, n *influxql.BinaryExp
}
// For fields, return all series from this measurement.
if key.Val != "_name" && ((key.Type == influxql.Unknown && mf.HasField(key.Val)) || key.Type == influxql.AnyField || (key.Type != influxql.Tag && key.Type != influxql.Unknown)) {
if key.Val != "_name" && ((key.Type == influxql.Unknown && is.HasField(name, key.Val)) || key.Type == influxql.AnyField || (key.Type != influxql.Tag && key.Type != influxql.Unknown)) {
itr, err := is.measurementSeriesIDIterator(name)
if err != nil {
return nil, err
@ -1838,7 +1858,7 @@ func (is IndexSet) seriesByBinaryExprIterator(name []byte, n *influxql.BinaryExp
return newSeriesIDExprIterator(itr, n), nil
} else if value, ok := value.(*influxql.VarRef); ok {
// Check if the RHS is a variable and if it is a field.
if value.Val != "_name" && ((value.Type == influxql.Unknown && mf.HasField(value.Val)) || key.Type == influxql.AnyField || (value.Type != influxql.Tag && value.Type != influxql.Unknown)) {
if value.Val != "_name" && ((value.Type == influxql.Unknown && is.HasField(name, value.Val)) || key.Type == influxql.AnyField || (value.Type != influxql.Tag && value.Type != influxql.Unknown)) {
itr, err := is.measurementSeriesIDIterator(name)
if err != nil {
return nil, err
@ -2130,7 +2150,7 @@ func (is IndexSet) matchTagValueNotEqualNotEmptySeriesIDIterator(name, key []byt
func (is IndexSet) TagValuesByKeyAndExpr(auth query.Authorizer, name []byte, keys []string, expr influxql.Expr, fieldset *MeasurementFieldSet) ([]map[string]struct{}, error) {
release := is.SeriesFile.Retain()
defer release()
return is.tagValuesByKeyAndExpr(auth, name, keys, expr, fieldset)
return is.tagValuesByKeyAndExpr(auth, name, keys, expr)
}
// tagValuesByKeyAndExpr retrieves tag values for the provided tag keys. See
@ -2138,10 +2158,10 @@ func (is IndexSet) TagValuesByKeyAndExpr(auth query.Authorizer, name []byte, key
//
// tagValuesByKeyAndExpr guarantees to never take any locks on the underlying
// series file.
func (is IndexSet) tagValuesByKeyAndExpr(auth query.Authorizer, name []byte, keys []string, expr influxql.Expr, fieldset *MeasurementFieldSet) ([]map[string]struct{}, error) {
func (is IndexSet) tagValuesByKeyAndExpr(auth query.Authorizer, name []byte, keys []string, expr influxql.Expr) ([]map[string]struct{}, error) {
database := is.Database()
itr, err := is.seriesByExprIterator(name, expr, fieldset.Fields(string(name)))
itr, err := is.seriesByExprIterator(name, expr)
if err != nil {
return nil, err
} else if itr == nil {
@ -2290,7 +2310,7 @@ func (is IndexSet) MeasurementTagKeyValuesByExpr(auth query.Authorizer, name []b
// This is the case where we have filtered series by some WHERE condition.
// We only care about the tag values for the keys given the
// filtered set of series ids.
resultSet, err := is.tagValuesByKeyAndExpr(auth, name, keys, expr, is.FieldSet())
resultSet, err := is.tagValuesByKeyAndExpr(auth, name, keys, expr)
if err != nil {
return nil, err
}

13
tsdb/shard.go
View File

@ -1405,6 +1405,9 @@ func (m *MeasurementFields) Field(name string) *Field {
}
func (m *MeasurementFields) HasField(name string) bool {
if m == nil {
return false
}
m.mu.RLock()
f := m.fields[name]
m.mu.RUnlock()
@ -1480,7 +1483,15 @@ func NewMeasurementFieldSet(path string) (*MeasurementFieldSet, error) {
}
// Fields returns fields for a measurement by name.
func (fs *MeasurementFieldSet) Fields(name string) *MeasurementFields {
func (fs *MeasurementFieldSet) Fields(name []byte) *MeasurementFields {
fs.mu.RLock()
mf := fs.fields[string(name)]
fs.mu.RUnlock()
return mf
}
// FieldsByString returns fields for a measurment by name.
func (fs *MeasurementFieldSet) FieldsByString(name string) *MeasurementFields {
fs.mu.RLock()
mf := fs.fields[name]
fs.mu.RUnlock()

6
tsdb/shard_test.go
View File

@ -1548,7 +1548,7 @@ func TestMeasurementFieldSet_SaveLoad(t *testing.T) {
t.Fatalf("NewMeasurementFieldSet error: %v", err)
}
fields = mf.Fields("cpu")
fields = mf.FieldsByString("cpu")
field := fields.Field("value")
if field == nil {
t.Fatalf("field is null")
@ -1593,7 +1593,7 @@ func TestMeasurementFieldSet_Corrupt(t *testing.T) {
t.Fatal("NewMeasurementFieldSet expected error")
}
fields = mf.Fields("cpu")
fields = mf.FieldsByString("cpu")
if fields != nil {
t.Fatal("expecte fields to be nil")
}
@ -1622,7 +1622,7 @@ func TestMeasurementFieldSet_DeleteEmpty(t *testing.T) {
t.Fatalf("NewMeasurementFieldSet error: %v", err)
}
fields = mf.Fields("cpu")
fields = mf.FieldsByString("cpu")
field := fields.Field("value")
if field == nil {
t.Fatalf("field is null")

23
tsdb/store_test.go
View File

@ -1138,18 +1138,24 @@ func TestStore_TagValues(t *testing.T) {
Name: "No WHERE clause",
Expr: &base,
Exp: []tsdb.TagValues{
createTagValues("cpu0", map[string][]string{"host": {"nofoo", "tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu1", map[string][]string{"host": {"nofoo", "tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu2", map[string][]string{"host": {"nofoo", "tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu0", map[string][]string{"shard": {"s0"}}),
createTagValues("cpu1", map[string][]string{"shard": {"s1"}}),
createTagValues("cpu10", map[string][]string{"host": {"nofoo", "tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu11", map[string][]string{"host": {"nofoo", "tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu12", map[string][]string{"host": {"nofoo", "tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu2", map[string][]string{"shard": {"s2"}}),
},
},
{
Name: "With WHERE clause",
Expr: baseWhere,
Exp: []tsdb.TagValues{
createTagValues("cpu0", map[string][]string{"host": {"tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu1", map[string][]string{"host": {"tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu2", map[string][]string{"host": {"tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu0", map[string][]string{"shard": {"s0"}}),
createTagValues("cpu1", map[string][]string{"shard": {"s1"}}),
createTagValues("cpu10", map[string][]string{"host": {"tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu11", map[string][]string{"host": {"tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu12", map[string][]string{"host": {"tv0", "tv1", "tv2", "tv3"}, "shard": {"s0", "s1", "s2"}}),
createTagValues("cpu2", map[string][]string{"shard": {"s2"}}),
},
},
}
@ -1158,9 +1164,10 @@ func TestStore_TagValues(t *testing.T) {
setup := func(index string) []uint64 { // returns shard ids
s = MustOpenStore(index)
fmtStr := `cpu%[1]d,foo=a,ignoreme=nope,host=tv%[2]d,shard=s%[3]d value=1 %[4]d
cpu%[1]d,host=nofoo value=1 %[4]d
fmtStr := `cpu1%[1]d,foo=a,ignoreme=nope,host=tv%[2]d,shard=s%[3]d value=1 %[4]d
cpu1%[1]d,host=nofoo value=1 %[4]d
mem,host=nothanks value=1 %[4]d
cpu%[3]d,shard=s%[3]d,foo=a value=2 %[4]d
`
genPoints := func(sid int) []string {
var ts int