Prevent panic in DecodeSameTypeBlock

If DecodeSameTypeBlock is called on on an empty Values slice, it would
panic with an index out of bounds error.  This func can actually be removed
because DecodeBlock can determine what type of values are encoded already.

This will still panic if the block cannot be decoded due to other reasons.

Fixes #4365

CHANGELOG.md

@@ -49,6 +49,7 @@
 - [#3429](https://github.com/influxdb/influxdb/issues/3429): Incorrect parsing of regex containing '/'
 - [#4374](https://github.com/influxdb/influxdb/issues/4374): Add tsm1 quickcheck tests
 - [#4377](https://github.com/influxdb/influxdb/pull/4377): Hinted handoff should not process dropped nodes
+- [#4365](https://github.com/influxdb/influxdb/issues/4365): Prevent panic in DecodeSameTypeBlock
 
 ## v0.9.4 [2015-09-14]
 

tsdb/engine/tsm1/encoding.go

@@ -69,7 +69,13 @@ func (v Values) MaxTime() int64 {
 	return v[len(v)-1].Time().UnixNano()
 }
 
+// Encode converts the values to a byte slice.  If there are no values,
+// this function panics.
 func (v Values) Encode(buf []byte) ([]byte, error) {
+	if len(v) == 0 {
+		panic("unable to encode block type")
+	}
+
 	switch v[0].(type) {
 	case *FloatValue:
 		return encodeFloatBlock(buf, v)
@@ -84,24 +90,6 @@ func (v Values) Encode(buf []byte) ([]byte, error) {
 	return nil, fmt.Errorf("unsupported value type %T", v[0])
 }
 
-func (v Values) DecodeSameTypeBlock(block []byte) Values {
-	switch v[0].(type) {
-	case *FloatValue:
-		a, _ := decodeFloatBlock(block)
-		return a
-	case *Int64Value:
-		a, _ := decodeInt64Block(block)
-		return a
-	case *BoolValue:
-		a, _ := decodeBoolBlock(block)
-		return a
-	case *StringValue:
-		a, _ := decodeStringBlock(block)
-		return a
-	}
-	return nil
-}
-
 // DecodeBlock takes a byte array and will decode into values of the appropriate type
 // based on the block
 func DecodeBlock(block []byte) (Values, error) {

tsdb/engine/tsm1/encoding_test.go

@@ -24,7 +24,10 @@ func TestEncoding_FloatBlock(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	decodedValues := values.DecodeSameTypeBlock(b)
+	decodedValues, err := tsm1.DecodeBlock(b)
+	if err != nil {
+		t.Fatalf("unexpected error decoding block: %v", err)
+	}
 
 	if !reflect.DeepEqual(decodedValues, values) {
 		t.Fatalf("unexpected results:\n\tgot: %v\n\texp: %v\n", decodedValues, values)
@@ -42,7 +45,10 @@ func TestEncoding_FloatBlock_ZeroTime(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	decodedValues := values.DecodeSameTypeBlock(b)
+	decodedValues, err := tsm1.DecodeBlock(b)
+	if err != nil {
+		t.Fatalf("unexpected error decoding block: %v", err)
+	}
 
 	if !reflect.DeepEqual(decodedValues, values) {
 		t.Fatalf("unexpected results:\n\tgot: %v\n\texp: %v\n", decodedValues, values)
@@ -62,7 +68,10 @@ func TestEncoding_FloatBlock_SimilarFloats(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	decodedValues := values.DecodeSameTypeBlock(b)
+	decodedValues, err := tsm1.DecodeBlock(b)
+	if err != nil {
+		t.Fatalf("unexpected error decoding block: %v", err)
+	}
 
 	if !reflect.DeepEqual(decodedValues, values) {
 		t.Fatalf("unexpected results:\n\tgot: %v\n\texp: %v\n", decodedValues, values)
@@ -82,7 +91,10 @@ func TestEncoding_IntBlock_Basic(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	decodedValues := values.DecodeSameTypeBlock(b)
+	decodedValues, err := tsm1.DecodeBlock(b)
+	if err != nil {
+		t.Fatalf("unexpected error decoding block: %v", err)
+	}
 
 	if len(decodedValues) != len(values) {
 		t.Fatalf("unexpected results length:\n\tgot: %v\n\texp: %v\n", len(decodedValues), len(values))
@@ -117,7 +129,10 @@ func TestEncoding_IntBlock_Negatives(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	decodedValues := values.DecodeSameTypeBlock(b)
+	decodedValues, err := tsm1.DecodeBlock(b)
+	if err != nil {
+		t.Fatalf("unexpected error decoding block: %v", err)
+	}
 
 	if !reflect.DeepEqual(decodedValues, values) {
 		t.Fatalf("unexpected results:\n\tgot: %v\n\texp: %v\n", decodedValues, values)
@@ -141,7 +156,10 @@ func TestEncoding_BoolBlock_Basic(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	decodedValues := values.DecodeSameTypeBlock(b)
+	decodedValues, err := tsm1.DecodeBlock(b)
+	if err != nil {
+		t.Fatalf("unexpected error decoding block: %v", err)
+	}
 
 	if !reflect.DeepEqual(decodedValues, values) {
 		t.Fatalf("unexpected results:\n\tgot: %v\n\texp: %v\n", decodedValues, values)
@@ -161,7 +179,10 @@ func TestEncoding_StringBlock_Basic(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	decodedValues := values.DecodeSameTypeBlock(b)
+	decodedValues, err := tsm1.DecodeBlock(b)
+	if err != nil {
+		t.Fatalf("unexpected error decoding block: %v", err)
+	}
 
 	if !reflect.DeepEqual(decodedValues, values) {
 		t.Fatalf("unexpected results:\n\tgot: %v\n\texp: %v\n", decodedValues, values)

tsdb/engine/tsm1/tsm1.go

@@ -614,7 +614,11 @@ func (e *Engine) Compact(fullCompaction bool) error {
 			for {
 				// write the values, the block or combine with previous
 				if len(previousValues) > 0 {
-					previousValues = append(previousValues, previousValues.DecodeSameTypeBlock(block)...)
+					decoded, err := DecodeBlock(block)
+					if err != nil {
+						panic(fmt.Sprintf("failure decoding block: %v", err))
+					}
+					previousValues = append(previousValues, decoded...)
 				} else if len(block) > e.RotateBlockSize {
 					if _, err := f.Write(df.mmap[pos:newPos]); err != nil {
 						return err
@@ -1645,7 +1649,10 @@ func (e *Engine) readSeries() (map[string]*tsdb.Series, error) {
 // has future encoded blocks so that this method can know how much of its values can be
 // combined and output in the resulting encoded block.
 func (e *Engine) DecodeAndCombine(newValues Values, block, buf []byte, nextTime int64, hasFutureBlock bool) (Values, []byte, error) {
-	values := newValues.DecodeSameTypeBlock(block)
+	values, err := DecodeBlock(block)
+	if err != nil {
+		panic(fmt.Sprintf("failure decoding block: %v", err))
+	}
 
 	var remainingValues Values