fix(storage): array cursor iterator should return stats of all observed cursors

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v2.0.0-alpha.20 [unreleased]
+
+### Bug Fixes
+
+1. [15731](https://github.com/influxdata/influxdb/pull/15731): Ensure array cursor iterator stats accumulate all cursor stats
+
 ## v2.0.0-alpha.19 [2019-10-30]
 
 ### Features

tsdb/tsm1/array_cursor_iterator.go

@@ -79,24 +79,32 @@ func (q *arrayCursorIterator) Stats() cursors.CursorStats {
 	var stats cursors.CursorStats
 	if cur := q.asc.Float; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.asc.Integer; cur != nil {
+	}
+	if cur := q.asc.Integer; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.asc.Unsigned; cur != nil {
+	}
+	if cur := q.asc.Unsigned; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.asc.Boolean; cur != nil {
+	}
+	if cur := q.asc.Boolean; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.asc.String; cur != nil {
+	}
+	if cur := q.asc.String; cur != nil {
 		stats.Add(cur.Stats())
 	}
 	if cur := q.desc.Float; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.desc.Integer; cur != nil {
+	}
+	if cur := q.desc.Integer; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.desc.Unsigned; cur != nil {
+	}
+	if cur := q.desc.Unsigned; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.desc.Boolean; cur != nil {
+	}
+	if cur := q.desc.Boolean; cur != nil {
 		stats.Add(cur.Stats())
-	} else if cur := q.desc.String; cur != nil {
+	}
+	if cur := q.desc.String; cur != nil {
 		stats.Add(cur.Stats())
 	}
 	return stats

tsdb/tsm1/engine_cursor_test.go

@@ -0,0 +1,117 @@
+package tsm1_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/influxdata/influxdb/models"
+	"github.com/influxdata/influxdb/tsdb"
+	"github.com/influxdata/influxdb/tsdb/cursors"
+)
+
+func TestEngine_CursorIterator_Stats(t *testing.T) {
+	e := MustOpenEngine()
+	defer e.Close()
+
+	points := []models.Point{
+		models.MustNewPoint("cpu",
+			models.Tags{
+				{Key: []byte("a"), Value: []byte("b")},
+			},
+			models.Fields{"value": 4.6},
+			time.Now().UTC(),
+		),
+		models.MustNewPoint("cpu",
+			models.Tags{
+				{Key: []byte("a"), Value: []byte("b")},
+			},
+			models.Fields{"value": 3.2},
+			time.Now().UTC(),
+		),
+		models.MustNewPoint("mem",
+			models.Tags{
+				{Key: []byte("b"), Value: []byte("c")},
+			},
+			models.Fields{"value": int64(3)},
+			time.Now().UTC(),
+		),
+	}
+
+	// Write into the index.
+	collection := tsdb.NewSeriesCollection(points)
+	if err := e.index.CreateSeriesListIfNotExists(collection); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := e.WritePoints(points); err != nil {
+		t.Fatal(err)
+	}
+
+	e.MustWriteSnapshot()
+
+	ctx := context.Background()
+	cursorIterator, err := e.CreateCursorIterator(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cur, err := cursorIterator.Next(ctx, &tsdb.CursorRequest{
+		Name:      []byte("cpu"),
+		Tags:      []models.Tag{{Key: []byte("a"), Value: []byte("b")}},
+		Field:     "value",
+		EndTime:   time.Now().UTC().UnixNano(),
+		Ascending: true,
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if cur == nil {
+		t.Fatal("expected cursor to be present")
+	}
+
+	fc, ok := cur.(cursors.FloatArrayCursor)
+	if !ok {
+		t.Fatalf("unexpected cursor type: expected FloatArrayCursor, got %#v", cur)
+	}
+
+	// drain the cursor
+	for a := fc.Next(); a.Len() > 0; a = fc.Next() {
+	}
+
+	cur.Close()
+
+	cur, err = cursorIterator.Next(ctx, &tsdb.CursorRequest{
+		Name:      []byte("mem"),
+		Tags:      []models.Tag{{Key: []byte("b"), Value: []byte("c")}},
+		Field:     "value",
+		EndTime:   time.Now().UTC().UnixNano(),
+		Ascending: true,
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if cur == nil {
+		t.Fatal("expected cursor to be present")
+	}
+
+	defer cur.Close()
+
+	ic, ok := cur.(cursors.IntegerArrayCursor)
+	if !ok {
+		t.Fatalf("unexpected cursor type: expected FloatArrayCursor, got %#v", cur)
+	}
+
+	// drain the cursor
+	for a := ic.Next(); a.Len() > 0; a = ic.Next() {
+	}
+
+	// iterator should report integer array stats
+	if got, exp := cursorIterator.Stats(), (cursors.CursorStats{ScannedValues: 3, ScannedBytes: 24}); exp != got {
+		t.Fatalf("expected %v, got %v", exp, got)
+	}
+}