fix: Bubble up errors for cursors

devanbenz · devanbenz · commit c8ee8127afc2 · 2025-03-20T11:42:06.000-05:00
diff --git a/mock/reads_resultset.go b/mock/reads_resultset.go
@@ -1,6 +1,8 @@
 package mock
 
 import (
+	"fmt"
+
 	"github.com/influxdata/influxdb/v2/models"
 	"github.com/influxdata/influxdb/v2/pkg/data/gen"
 	"github.com/influxdata/influxdb/v2/storage/reads"
@@ -61,7 +63,7 @@ func (g *GeneratorResultSet) Next() bool {
 	return g.sg.Next() && (g.max == 0 || remain > 0)
 }
 
-func (g *GeneratorResultSet) Cursor() cursors.Cursor {
+func (g *GeneratorResultSet) Cursor() (cursors.Cursor, error) {
 	switch g.sg.FieldType() {
 	case models.Float:
 		g.f.tv = g.sg.TimeValuesGenerator()
@@ -79,10 +81,10 @@ func (g *GeneratorResultSet) Cursor() cursors.Cursor {
 		g.b.tv = g.sg.TimeValuesGenerator()
 		g.cur = &g.b
 	default:
-		panic("unreachable")
+		return nil, fmt.Errorf("unsupported field type: %v", g.sg.FieldType())
 	}
 
-	return g.cur
+	return g.cur, nil
 }
 
 func copyTags(dst, src models.Tags) models.Tags {
diff --git a/storage/flux/reader.go b/storage/flux/reader.go
@@ -194,7 +194,10 @@ func (fi *filterIterator) handleRead(f func(flux.Table) error, rs storage.Result
 
 READ:
 	for rs.Next() {
-		cur = rs.Cursor()
+		cur, err := rs.Cursor()
+		if err != nil {
+			return err
+		}
 		if cur == nil {
 			// no data for series key + field combination
 			continue
@@ -331,7 +334,7 @@ func (gi *groupIterator) handleRead(f func(flux.Table) error, rs storage.GroupRe
 READ:
 	for gc != nil {
 		for gc.Next() {
-			cur = gc.Cursor()
+			cur, _ = gc.Cursor()
 			if cur != nil {
 				break
 			}
@@ -740,7 +743,10 @@ func (wai *windowAggregateIterator) handleRead(f func(flux.Table) error, rs stor
 
 READ:
 	for rs.Next() {
-		cur = rs.Cursor()
+		cur, err = rs.Cursor()
+		if err != nil {
+			return err
+		}
 		if cur == nil {
 			// no data for series key + field combination
 			continue
diff --git a/storage/flux/table.gen.go b/storage/flux/table.gen.go
@@ -970,7 +970,7 @@ func (t *floatGroupTable) advanceCursor() bool {
 	t.cur.Close()
 	t.cur = nil
 	for t.gc.Next() {
-		cur := t.gc.Cursor()
+		cur, _ := t.gc.Cursor()
 		if cur == nil {
 			continue
 		}
@@ -1954,7 +1954,7 @@ func (t *integerGroupTable) advanceCursor() bool {
 	t.cur.Close()
 	t.cur = nil
 	for t.gc.Next() {
-		cur := t.gc.Cursor()
+		cur, _ := t.gc.Cursor()
 		if cur == nil {
 			continue
 		}
@@ -2935,7 +2935,7 @@ func (t *unsignedGroupTable) advanceCursor() bool {
 	t.cur.Close()
 	t.cur = nil
 	for t.gc.Next() {
-		cur := t.gc.Cursor()
+		cur, _ := t.gc.Cursor()
 		if cur == nil {
 			continue
 		}
@@ -3860,7 +3860,7 @@ func (t *stringGroupTable) advanceCursor() bool {
 	t.cur.Close()
 	t.cur = nil
 	for t.gc.Next() {
-		cur := t.gc.Cursor()
+		cur, _ := t.gc.Cursor()
 		if cur == nil {
 			continue
 		}
@@ -4785,7 +4785,7 @@ func (t *booleanGroupTable) advanceCursor() bool {
 	t.cur.Close()
 	t.cur = nil
 	for t.gc.Next() {
-		cur := t.gc.Cursor()
+		cur, _ := t.gc.Cursor()
 		if cur == nil {
 			continue
 		}
diff --git a/storage/flux/table.gen.go.tmpl b/storage/flux/table.gen.go.tmpl
@@ -1000,7 +1000,7 @@ func (t *{{.name}}GroupTable) advanceCursor() bool {
 	t.cur.Close()
 	t.cur = nil
 	for t.gc.Next() {
-		cur := t.gc.Cursor()
+		cur, _ := t.gc.Cursor()
 		if cur == nil {
 			continue
 		}
diff --git a/storage/reads/aggregate_resultset.go b/storage/reads/aggregate_resultset.go
@@ -97,7 +97,10 @@ func (r *windowAggregateResultSet) createCursor(seriesRow SeriesRow) (cursors.Cu
 	agg := r.req.Aggregate[0]
 	every := r.req.WindowEvery
 	offset := r.req.Offset
-	cursor := r.arrayCursors.createCursor(seriesRow)
+	cursor, err := r.arrayCursors.createCursor(seriesRow)
+	if err != nil {
+		return nil, err
+	}
 
 	var everyDur values.Duration
 	var offsetDur values.Duration
@@ -132,8 +135,8 @@ func (r *windowAggregateResultSet) createCursor(seriesRow SeriesRow) (cursors.Cu
 	}
 }
 
-func (r *windowAggregateResultSet) Cursor() cursors.Cursor {
-	return r.cursor
+func (r *windowAggregateResultSet) Cursor() (cursors.Cursor, error) {
+	return r.cursor, nil
 }
 
 func (r *windowAggregateResultSet) Close() {
diff --git a/storage/reads/aggregate_resultset_test.go b/storage/reads/aggregate_resultset_test.go
@@ -2,6 +2,7 @@ package reads_test
 
 import (
 	"context"
+	"github.com/stretchr/testify/require"
 	"reflect"
 	"testing"
 	"time"
@@ -196,7 +197,8 @@ func TestNewWindowAggregateResultSet_Mean(t *testing.T) {
 	if !resultSet.Next() {
 		t.Fatalf("unexpected: resultSet could not advance")
 	}
-	cursor := resultSet.Cursor()
+	cursor, err := resultSet.Cursor()
+	require.NoError(t, err, "create cursor failed")
 	if cursor == nil {
 		t.Fatalf("unexpected: cursor was nil")
 	}
@@ -238,7 +240,8 @@ func TestNewWindowAggregateResultSet_Months(t *testing.T) {
 	if !resultSet.Next() {
 		t.Fatalf("unexpected: resultSet could not advance")
 	}
-	cursor := resultSet.Cursor()
+	cursor, err := resultSet.Cursor()
+	require.NoError(t, err, "create cursor failed")
 	if cursor == nil {
 		t.Fatalf("unexpected: cursor was nil")
 	}
diff --git a/storage/reads/array_cursor.gen.go b/storage/reads/array_cursor.gen.go
@@ -45,7 +45,7 @@ func newLimitArrayCursor(cur cursors.Cursor) (cursors.Cursor, error) {
 	default:
 		return nil, &errors2.Error{
 			Code: errors2.EInvalid,
-			Msg:  fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),
+			Msg:  fmt.Sprintf("unsupported limit array cursor type: %s", arrayCursorType(cur)),
 		}
 	}
 }
@@ -74,7 +74,7 @@ func newWindowFirstArrayCursor(cur cursors.Cursor, window interval.Window) (curs
 	default:
 		return nil, &errors2.Error{
 			Code: errors2.EInvalid,
-			Msg:  fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),
+			Msg:  fmt.Sprintf("unsupported window first cursor type: %s", arrayCursorType(cur)),
 		}
 	}
 }
@@ -103,7 +103,7 @@ func newWindowLastArrayCursor(cur cursors.Cursor, window interval.Window) (curso
 	default:
 		return nil, &errors2.Error{
 			Code: errors2.EInvalid,
-			Msg:  fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),
+			Msg:  fmt.Sprintf("unsupported window last cursor type: %s", arrayCursorType(cur)),
 		}
 	}
 }
@@ -129,7 +129,7 @@ func newWindowCountArrayCursor(cur cursors.Cursor, window interval.Window) (curs
 	default:
 		return nil, &errors2.Error{
 			Code: errors2.EInvalid,
-			Msg:  fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),
+			Msg:  fmt.Sprintf("unsupported window count cursor type: %s", arrayCursorType(cur)),
 		}
 	}
 }
diff --git a/storage/reads/array_cursor.go b/storage/reads/array_cursor.go
@@ -3,6 +3,7 @@ package reads
 import (
 	"context"
 	"fmt"
+
 	errors2 "github.com/influxdata/influxdb/v2/kit/platform/errors"
 
 	"github.com/influxdata/flux/interval"
@@ -104,7 +105,7 @@ func newMultiShardArrayCursors(ctx context.Context, start, end int64, asc bool)
 	return m
 }
 
-func (m *multiShardArrayCursors) createCursor(row SeriesRow) cursors.Cursor {
+func (m *multiShardArrayCursors) createCursor(row SeriesRow) (cursors.Cursor, error) {
 	m.req.Name = row.Name
 	m.req.Tags = row.SeriesTags
 	m.req.Field = row.Field
@@ -123,26 +124,29 @@ func (m *multiShardArrayCursors) createCursor(row SeriesRow) cursors.Cursor {
 	}
 
 	if cur == nil || err != nil {
-		return nil
+		return nil, err
 	}
 
 	switch c := cur.(type) {
 	case cursors.IntegerArrayCursor:
 		m.cursors.i.reset(c, row.Query, cond)
-		return &m.cursors.i
+		return &m.cursors.i, nil
 	case cursors.FloatArrayCursor:
 		m.cursors.f.reset(c, row.Query, cond)
-		return &m.cursors.f
+		return &m.cursors.f, nil
 	case cursors.UnsignedArrayCursor:
 		m.cursors.u.reset(c, row.Query, cond)
-		return &m.cursors.u
+		return &m.cursors.u, nil
 	case cursors.StringArrayCursor:
 		m.cursors.s.reset(c, row.Query, cond)
-		return &m.cursors.s
+		return &m.cursors.s, nil
 	case cursors.BooleanArrayCursor:
 		m.cursors.b.reset(c, row.Query, cond)
-		return &m.cursors.b
+		return &m.cursors.b, nil
 	default:
-		return nil
+		return nil, &errors2.Error{
+			Code: errors2.EInvalid,
+			Msg:  fmt.Sprintf("unsupported cursor type: %s", arrayCursorType(cur)),
+		}
 	}
 }
diff --git a/storage/reads/array_cursor_test.go b/storage/reads/array_cursor_test.go
@@ -2227,7 +2227,9 @@ func TestMultiShardArrayCursor(t *testing.T) {
 		row := SeriesRow{Query: iter}
 		ctx := context.Background()
 		msac := newMultiShardArrayCursors(ctx, models.MinNanoTime, models.MaxNanoTime, true)
-		cur, ok := msac.createCursor(row).(cursors.IntegerArrayCursor)
+		cursor, err := msac.createCursor(row)
+		require.NoError(t, err, "create cursor failed")
+		cur, ok := cursor.(cursors.IntegerArrayCursor)
 		require.Truef(t, ok, "Expected IntegerArrayCursor")
 
 		ia := cur.Next()
diff --git a/storage/reads/group_resultset.go b/storage/reads/group_resultset.go
@@ -121,7 +121,10 @@ func (g *groupResultSet) Next() GroupCursor {
 // the time range of the query.
 func (g *groupResultSet) seriesHasPoints(row *SeriesRow) bool {
 	// TODO(sgc): this is expensive. Storage engine must provide efficient time range queries of series keys.
-	cur := g.arrayCursors.createCursor(*row)
+	cur, err := g.arrayCursors.createCursor(*row)
+	if err != nil {
+		return false
+	}
 	var ts []int64
 	switch c := cur.(type) {
 	case cursors.IntegerArrayCursor:
@@ -302,15 +305,18 @@ func (c *groupNoneCursor) Next() bool {
 }
 
 func (c *groupNoneCursor) createCursor(seriesRow SeriesRow) (cur cursors.Cursor, err error) {
-	cur = c.arrayCursors.createCursor(c.row)
+	cur, err = c.arrayCursors.createCursor(c.row)
+	if err != nil {
+		return nil, err
+	}
 	if c.agg != nil {
 		cur, err = newAggregateArrayCursor(c.ctx, c.agg, cur)
 	}
 	return cur, err
 }
 
-func (c *groupNoneCursor) Cursor() cursors.Cursor {
-	return c.cursor
+func (c *groupNoneCursor) Cursor() (cursors.Cursor, error) {
+	return c.cursor, nil
 }
 
 type groupByCursor struct {
@@ -350,15 +356,18 @@ func (c *groupByCursor) Next() bool {
 }
 
 func (c *groupByCursor) createCursor(seriesRow SeriesRow) (cur cursors.Cursor, err error) {
-	cur = c.arrayCursors.createCursor(seriesRow)
+	cur, err = c.arrayCursors.createCursor(seriesRow)
+	if err != nil {
+		return nil, err
+	}
 	if c.agg != nil {
 		cur, err = newAggregateArrayCursor(c.ctx, c.agg, cur)
 	}
 	return cur, err
 }
 
-func (c *groupByCursor) Cursor() cursors.Cursor {
-	return c.cursor
+func (c *groupByCursor) Cursor() (cursors.Cursor, error) {
+	return c.cursor, nil
 }
 
 func (c *groupByCursor) Stats() cursors.CursorStats {
diff --git a/storage/reads/resultset.go b/storage/reads/resultset.go
@@ -8,7 +8,7 @@ import (
 )
 
 type multiShardCursors interface {
-	createCursor(row SeriesRow) cursors.Cursor
+	createCursor(row SeriesRow) (cursors.Cursor, error)
 }
 
 type resultSet struct {
@@ -57,7 +57,7 @@ func (r *resultSet) Next() bool {
 	return true
 }
 
-func (r *resultSet) Cursor() cursors.Cursor {
+func (r *resultSet) Cursor() (cursors.Cursor, error) {
 	return r.arrayCursors.createCursor(r.seriesRow)
 }
 
diff --git a/storage/reads/resultset_lineprotocol.go b/storage/reads/resultset_lineprotocol.go
@@ -32,7 +32,11 @@ func ResultSetToLineProtocol(wr io.Writer, rs ResultSet) (err error) {
 		line = append(line, ' ')
 		line = append(line, field...)
 		line = append(line, '=')
-		err = cursorToLineProtocol(wr, line, rs.Cursor())
+		cursor, err := rs.Cursor()
+		if err != nil {
+			return err
+		}
+		err = cursorToLineProtocol(wr, line, cursor)
 		if err != nil {
 			return err
 		}
diff --git a/storage/reads/store.go b/storage/reads/store.go
@@ -15,7 +15,7 @@ type ResultSet interface {
 	Next() bool
 
 	// Cursor returns the most recent cursor after a call to Next.
-	Cursor() cursors.Cursor
+	Cursor() (cursors.Cursor, error)
 
 	// Tags returns the tags for the most recent cursor after a call to Next.
 	Tags() models.Tags
@@ -47,7 +47,7 @@ type GroupCursor interface {
 	Next() bool
 
 	// Cursor returns the most recent cursor after a call to Next.
-	Cursor() cursors.Cursor
+	Cursor() (cursors.Cursor, error)
 
 	// Tags returns the tags for the most recent cursor after a call to Next.
 	Tags() models.Tags
diff --git a/storage/reads/store_test.go b/storage/reads/store_test.go
@@ -119,7 +119,7 @@ func resultSetToString(wr io.Writer, rs reads.ResultSet, opts ...optionFn) {
 	for rs.Next() {
 		fmt.Fprint(wr, "series: ")
 		tagsToString(wr, rs.Tags())
-		cur := rs.Cursor()
+		cur, _ := rs.Cursor()
 
 		if po.SkipNilCursor && cur == nil {
 			continue
diff --git a/v1/services/storage/store.go b/v1/services/storage/store.go

Original file line number	Diff line number	Diff line change
`@@ -1000,7 +1000,7 @@ func (t *{{.name}}GroupTable) advanceCursor() bool {`
`1000`	`1000`	`t.cur.Close()`
`1001`	`1001`	`t.cur = nil`
`1002`	`1002`	`for t.gc.Next() {`
`1003`		`- cur := t.gc.Cursor()`
	`1003`	`+ cur, _ := t.gc.Cursor()`
`1004`	`1004`	`if cur == nil {`
`1005`	`1005`	`continue`
`1006`	`1006`	`}`
Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,10 @@ func (r *windowAggregateResultSet) createCursor(seriesRow SeriesRow) (cursors.Cu`
`97`	`97`	`agg := r.req.Aggregate[0]`
`98`	`98`	`every := r.req.WindowEvery`
`99`	`99`	`offset := r.req.Offset`
`100`		`- cursor := r.arrayCursors.createCursor(seriesRow)`
	`100`	`+ cursor, err := r.arrayCursors.createCursor(seriesRow)`
	`101`	`+ if err != nil {`
	`102`	`+ return nil, err`
	`103`	`+ }`
`101`	`104`
`102`	`105`	`var everyDur values.Duration`
`103`	`106`	`var offsetDur values.Duration`
`@@ -132,8 +135,8 @@ func (r *windowAggregateResultSet) createCursor(seriesRow SeriesRow) (cursors.Cu`
`132`	`135`	`}`
`133`	`136`	`}`
`134`	`137`
`135`		`-func (r *windowAggregateResultSet) Cursor() cursors.Cursor {`
`136`		`- return r.cursor`
	`138`	`+func (r *windowAggregateResultSet) Cursor() (cursors.Cursor, error) {`
	`139`	`+ return r.cursor, nil`
`137`	`140`	`}`
`138`	`141`
`139`	`142`	`func (r *windowAggregateResultSet) Close() {`
Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,7 @@ func newLimitArrayCursor(cur cursors.Cursor) (cursors.Cursor, error) {`
`45`	`45`	`default:`
`46`	`46`	`return nil, &errors2.Error{`
`47`	`47`	`Code: errors2.EInvalid,`
`48`		`- Msg: fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),`
	`48`	`+ Msg: fmt.Sprintf("unsupported limit array cursor type: %s", arrayCursorType(cur)),`
`49`	`49`	`}`
`50`	`50`	`}`
`51`	`51`	`}`
`@@ -74,7 +74,7 @@ func newWindowFirstArrayCursor(cur cursors.Cursor, window interval.Window) (curs`
`74`	`74`	`default:`
`75`	`75`	`return nil, &errors2.Error{`
`76`	`76`	`Code: errors2.EInvalid,`
`77`		`- Msg: fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),`
	`77`	`+ Msg: fmt.Sprintf("unsupported window first cursor type: %s", arrayCursorType(cur)),`
`78`	`78`	`}`
`79`	`79`	`}`
`80`	`80`	`}`
`@@ -103,7 +103,7 @@ func newWindowLastArrayCursor(cur cursors.Cursor, window interval.Window) (curso`
`103`	`103`	`default:`
`104`	`104`	`return nil, &errors2.Error{`
`105`	`105`	`Code: errors2.EInvalid,`
`106`		`- Msg: fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),`
	`106`	`+ Msg: fmt.Sprintf("unsupported window last cursor type: %s", arrayCursorType(cur)),`
`107`	`107`	`}`
`108`	`108`	`}`
`109`	`109`	`}`
`@@ -129,7 +129,7 @@ func newWindowCountArrayCursor(cur cursors.Cursor, window interval.Window) (curs`
`129`	`129`	`default:`
`130`	`130`	`return nil, &errors2.Error{`
`131`	`131`	`Code: errors2.EInvalid,`
`132`		`- Msg: fmt.Sprintf("unreachable: %s", arrayCursorType(cur)),`
	`132`	`+ Msg: fmt.Sprintf("unsupported window count cursor type: %s", arrayCursorType(cur)),`
`133`	`133`	`}`
`134`	`134`	`}`
`135`	`135`	`}`