Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: return and respect cursor errors (#24791) #24846

Merged
merged 2 commits into from
Mar 26, 2024
Merged

fix: return and respect cursor errors (#24791) #24846

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2f7b56b
fix: return and respect cursor errors (#24791)
davidby-influx Mar 26, 2024
dfa5ea3
fix: MustTempDir and arg to NewFileStore added
davidby-influx Mar 26, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 25 additions & 15 deletions storage/reads/array_cursor.gen.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,13 +315,15 @@ func (c *floatMultiShardArrayCursor) nextArrayCursor() bool {

var itr cursors.CursorIterator
var cur cursors.Cursor
for cur == nil && len(c.itrs) > 0 {
var err error
for cur == nil && len(c.itrs) > 0 && err == nil {
itr, c.itrs = c.itrs[0], c.itrs[1:]
cur, _ = itr.Next(c.ctx, c.req)
cur, err = itr.Next(c.ctx, c.req)
}

c.err = err
var ok bool
if cur != nil {
if cur != nil && err == nil {
var next cursors.FloatArrayCursor
next, ok = cur.(cursors.FloatArrayCursor)
if !ok {
Expand Down Expand Up @@ -1196,13 +1198,15 @@ func (c *integerMultiShardArrayCursor) nextArrayCursor() bool {

var itr cursors.CursorIterator
var cur cursors.Cursor
for cur == nil && len(c.itrs) > 0 {
var err error
for cur == nil && len(c.itrs) > 0 && err == nil {
itr, c.itrs = c.itrs[0], c.itrs[1:]
cur, _ = itr.Next(c.ctx, c.req)
cur, err = itr.Next(c.ctx, c.req)
}

c.err = err
var ok bool
if cur != nil {
if cur != nil && err == nil {
var next cursors.IntegerArrayCursor
next, ok = cur.(cursors.IntegerArrayCursor)
if !ok {
Expand Down Expand Up @@ -2077,13 +2081,15 @@ func (c *unsignedMultiShardArrayCursor) nextArrayCursor() bool {

var itr cursors.CursorIterator
var cur cursors.Cursor
for cur == nil && len(c.itrs) > 0 {
var err error
for cur == nil && len(c.itrs) > 0 && err == nil {
itr, c.itrs = c.itrs[0], c.itrs[1:]
cur, _ = itr.Next(c.ctx, c.req)
cur, err = itr.Next(c.ctx, c.req)
}

c.err = err
var ok bool
if cur != nil {
if cur != nil && err == nil {
var next cursors.UnsignedArrayCursor
next, ok = cur.(cursors.UnsignedArrayCursor)
if !ok {
Expand Down Expand Up @@ -2958,13 +2964,15 @@ func (c *stringMultiShardArrayCursor) nextArrayCursor() bool {

var itr cursors.CursorIterator
var cur cursors.Cursor
for cur == nil && len(c.itrs) > 0 {
var err error
for cur == nil && len(c.itrs) > 0 && err == nil {
itr, c.itrs = c.itrs[0], c.itrs[1:]
cur, _ = itr.Next(c.ctx, c.req)
cur, err = itr.Next(c.ctx, c.req)
}

c.err = err
var ok bool
if cur != nil {
if cur != nil && err == nil {
var next cursors.StringArrayCursor
next, ok = cur.(cursors.StringArrayCursor)
if !ok {
Expand Down Expand Up @@ -3384,13 +3392,15 @@ func (c *booleanMultiShardArrayCursor) nextArrayCursor() bool {

var itr cursors.CursorIterator
var cur cursors.Cursor
for cur == nil && len(c.itrs) > 0 {
var err error
for cur == nil && len(c.itrs) > 0 && err == nil {
itr, c.itrs = c.itrs[0], c.itrs[1:]
cur, _ = itr.Next(c.ctx, c.req)
cur, err = itr.Next(c.ctx, c.req)
}

c.err = err
var ok bool
if cur != nil {
if cur != nil && err == nil {
var next cursors.BooleanArrayCursor
next, ok = cur.(cursors.BooleanArrayCursor)
if !ok {
Expand Down
10 changes: 6 additions & 4 deletions storage/reads/array_cursor.gen.go.tmpl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,13 +263,15 @@ func (c *{{.name}}MultiShardArrayCursor) nextArrayCursor() bool {

var itr cursors.CursorIterator
var cur cursors.Cursor
for cur == nil && len(c.itrs) > 0 {
var err error
for cur == nil && len(c.itrs) > 0 && err == nil {
itr, c.itrs = c.itrs[0], c.itrs[1:]
cur, _ = itr.Next(c.ctx, c.req)
}
cur, err = itr.Next(c.ctx, c.req)
}

c.err = err
var ok bool
if cur != nil {
if cur != nil && err == nil {
var next cursors.{{.Name}}ArrayCursor
next, ok = cur.(cursors.{{.Name}}ArrayCursor)
if !ok {
Expand Down
5 changes: 3 additions & 2 deletions storage/reads/array_cursor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,12 +113,13 @@ func (m *multiShardArrayCursors) createCursor(row SeriesRow) cursors.Cursor {

var shard cursors.CursorIterator
var cur cursors.Cursor
var err error
for cur == nil && len(row.Query) > 0 {
shard, row.Query = row.Query[0], row.Query[1:]
cur, _ = shard.Next(m.ctx, &m.req)
cur, err = shard.Next(m.ctx, &m.req)
}

if cur == nil {
if cur == nil || err != nil {
return nil
}

Expand Down
90 changes: 70 additions & 20 deletions tsdb/engine/tsm1/array_cursor.gen.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,18 +39,23 @@ func newFloatArrayAscendingCursor() *floatArrayAscendingCursor {
return c
}

func (c *floatArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *floatArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
c.end = end
c.cache.values = cacheValues
c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
return c.cache.values[i].UnixNano() >= seek
})

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadFloatArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadFloatArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] >= seek
})
return nil
}

func (c *floatArrayAscendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -182,7 +187,8 @@ func newFloatArrayDescendingCursor() *floatArrayDescendingCursor {
return c
}

func (c *floatArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *floatArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
// Search for the time value greater than the seek time (not included)
// and then move our position back one which will include the values in
// our time range.
Expand All @@ -194,11 +200,15 @@ func (c *floatArrayDescendingCursor) reset(seek, end int64, cacheValues Values,
c.cache.pos--

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadFloatArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadFloatArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] > seek
})
c.tsm.pos--
return nil
}

func (c *floatArrayDescendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -321,18 +331,23 @@ func newIntegerArrayAscendingCursor() *integerArrayAscendingCursor {
return c
}

func (c *integerArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *integerArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
c.end = end
c.cache.values = cacheValues
c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
return c.cache.values[i].UnixNano() >= seek
})

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadIntegerArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadIntegerArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] >= seek
})
return nil
}

func (c *integerArrayAscendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -464,7 +479,8 @@ func newIntegerArrayDescendingCursor() *integerArrayDescendingCursor {
return c
}

func (c *integerArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *integerArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
// Search for the time value greater than the seek time (not included)
// and then move our position back one which will include the values in
// our time range.
Expand All @@ -476,11 +492,15 @@ func (c *integerArrayDescendingCursor) reset(seek, end int64, cacheValues Values
c.cache.pos--

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadIntegerArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadIntegerArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] > seek
})
c.tsm.pos--
return nil
}

func (c *integerArrayDescendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -603,18 +623,23 @@ func newUnsignedArrayAscendingCursor() *unsignedArrayAscendingCursor {
return c
}

func (c *unsignedArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *unsignedArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
c.end = end
c.cache.values = cacheValues
c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
return c.cache.values[i].UnixNano() >= seek
})

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadUnsignedArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadUnsignedArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] >= seek
})
return nil
}

func (c *unsignedArrayAscendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -746,7 +771,8 @@ func newUnsignedArrayDescendingCursor() *unsignedArrayDescendingCursor {
return c
}

func (c *unsignedArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *unsignedArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
// Search for the time value greater than the seek time (not included)
// and then move our position back one which will include the values in
// our time range.
Expand All @@ -758,11 +784,15 @@ func (c *unsignedArrayDescendingCursor) reset(seek, end int64, cacheValues Value
c.cache.pos--

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadUnsignedArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadUnsignedArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] > seek
})
c.tsm.pos--
return nil
}

func (c *unsignedArrayDescendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -885,18 +915,23 @@ func newStringArrayAscendingCursor() *stringArrayAscendingCursor {
return c
}

func (c *stringArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *stringArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
c.end = end
c.cache.values = cacheValues
c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
return c.cache.values[i].UnixNano() >= seek
})

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadStringArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadStringArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] >= seek
})
return nil
}

func (c *stringArrayAscendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -1028,7 +1063,8 @@ func newStringArrayDescendingCursor() *stringArrayDescendingCursor {
return c
}

func (c *stringArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *stringArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
// Search for the time value greater than the seek time (not included)
// and then move our position back one which will include the values in
// our time range.
Expand All @@ -1040,11 +1076,15 @@ func (c *stringArrayDescendingCursor) reset(seek, end int64, cacheValues Values,
c.cache.pos--

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadStringArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadStringArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] > seek
})
c.tsm.pos--
return nil
}

func (c *stringArrayDescendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -1167,18 +1207,23 @@ func newBooleanArrayAscendingCursor() *booleanArrayAscendingCursor {
return c
}

func (c *booleanArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *booleanArrayAscendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
c.end = end
c.cache.values = cacheValues
c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
return c.cache.values[i].UnixNano() >= seek
})

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadBooleanArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadBooleanArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] >= seek
})
return nil
}

func (c *booleanArrayAscendingCursor) Err() error { return nil }
Expand Down Expand Up @@ -1310,7 +1355,8 @@ func newBooleanArrayDescendingCursor() *booleanArrayDescendingCursor {
return c
}

func (c *booleanArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) {
func (c *booleanArrayDescendingCursor) reset(seek, end int64, cacheValues Values, tsmKeyCursor *KeyCursor) error {
var err error
// Search for the time value greater than the seek time (not included)
// and then move our position back one which will include the values in
// our time range.
Expand All @@ -1322,11 +1368,15 @@ func (c *booleanArrayDescendingCursor) reset(seek, end int64, cacheValues Values
c.cache.pos--

c.tsm.keyCursor = tsmKeyCursor
c.tsm.values, _ = c.tsm.keyCursor.ReadBooleanArrayBlock(c.tsm.buf)
c.tsm.values, err = c.tsm.keyCursor.ReadBooleanArrayBlock(c.tsm.buf)
if err != nil {
return err
}
c.tsm.pos = sort.Search(c.tsm.values.Len(), func(i int) bool {
return c.tsm.values.Timestamps[i] > seek
})
c.tsm.pos--
return nil
}

func (c *booleanArrayDescendingCursor) Err() error { return nil }
Expand Down
Loading