fix(storage): assumulate stats in arrayCursorIterator.Stats() call ac…

…ross all observed cursors
influxdata · Nov 4, 2019 · 34ee47a · 34ee47a
1 parent 4af0c0a
commit 34ee47a
Showing 1 changed file with 37 additions and 70 deletions.
diff --git a/tsdb/tsm1/array_cursor_iterator.go b/tsdb/tsm1/array_cursor_iterator.go
@@ -12,10 +12,8 @@ import (
 )
 
 type arrayCursorIterator struct {
-	e     *Engine
-	key   []byte
-	id    tsdb.SeriesIDTyped
-	isAsc bool
+	e   *Engine
+	key []byte
 
 	asc struct {
 		Float    *floatArrayAscendingCursor
@@ -36,9 +34,8 @@ type arrayCursorIterator struct {
 
 func (q *arrayCursorIterator) Next(ctx context.Context, r *tsdb.CursorRequest) (tsdb.Cursor, error) {
 	q.key = tsdb.AppendSeriesKey(q.key[:0], r.Name, r.Tags)
-	q.isAsc = r.Ascending
-	q.id = q.e.sfile.SeriesIDTypedBySeriesKey(q.key)
-	if q.id.IsZero() {
+	id := q.e.sfile.SeriesIDTypedBySeriesKey(q.key)
+	if id.IsZero() {
 		return nil, nil
 	}
 
@@ -54,7 +51,7 @@ func (q *arrayCursorIterator) Next(ctx context.Context, r *tsdb.CursorRequest) (
 	opt.EndTime = r.EndTime
 
 	// Return appropriate cursor based on type.
-	switch typ := q.id.Type(); typ {
+	switch typ := id.Type(); typ {
 	case models.Float:
 		return q.buildFloatArrayCursor(ctx, r.Name, r.Tags, r.Field, opt), nil
 	case models.Integer:
@@ -78,67 +75,37 @@ func (q *arrayCursorIterator) seriesFieldKeyBytes(name []byte, tags models.Tags,
 }
 
 // Stats returns the cumulative stats for all cursors.
-func (q *arrayCursorIterator) Stats() (stats cursors.CursorStats) {
-	// Return appropriate cursor based on type.
-	switch typ := q.id.Type(); typ {
-	case models.Float:
-		if q.isAsc {
-			if cur := q.asc.Float; cur != nil {
-				stats.Add(cur.Stats())
-			}
-			return
-		}
-
-		if cur := q.desc.Float; cur != nil {
-			stats.Add(cur.Stats())
-		}
-	case models.Integer:
-		if q.isAsc {
-			if cur := q.asc.Integer; cur != nil {
-				stats.Add(cur.Stats())
-			}
-			return
-		}
-
-		if cur := q.desc.Integer; cur != nil {
-			stats.Add(cur.Stats())
-		}
-	case models.Unsigned:
-		if q.isAsc {
-			if cur := q.asc.Unsigned; cur != nil {
-				stats.Add(cur.Stats())
-			}
-			return
-		}
-
-		if cur := q.desc.Unsigned; cur != nil {
-			stats.Add(cur.Stats())
-		}
-	case models.String:
-		if q.isAsc {
-			if cur := q.asc.String; cur != nil {
-				stats.Add(cur.Stats())
-			}
-			return
-		}
-
-		if cur := q.desc.String; cur != nil {
-			stats.Add(cur.Stats())
-		}
-	case models.Boolean:
-		if q.isAsc {
-			if cur := q.asc.Boolean; cur != nil {
-				stats.Add(cur.Stats())
-			}
-			return
-		}
-
-		if cur := q.desc.Boolean; cur != nil {
-			stats.Add(cur.Stats())
-		}
-	default:
-		panic(fmt.Sprintf("unreachable: %v", typ))
+func (q *arrayCursorIterator) Stats() cursors.CursorStats {
+	var stats cursors.CursorStats
+	if cur := q.asc.Float; cur != nil {
+		stats.Add(cur.Stats())
 	}
-
-	return
+	if cur := q.asc.Integer; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.asc.Unsigned; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.asc.Boolean; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.asc.String; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.desc.Float; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.desc.Integer; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.desc.Unsigned; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.desc.Boolean; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	if cur := q.desc.String; cur != nil {
+		stats.Add(cur.Stats())
+	}
+	return stats
 }