db: propagate errors from scanInternalIterator close

Fix scanInternalIterator to no longer swallow errors on Close.

Author: jbowens

db.go

@@ -2696,8 +2696,7 @@ type ScanStatisticsOptions struct {
 // key span [lower, upper) as well as the number of snapshot keys.
 func (d *DB) ScanStatistics(
 	ctx context.Context, lower, upper []byte, opts ScanStatisticsOptions,
-) (LSMKeyStatistics, error) {
-	stats := LSMKeyStatistics{}
+) (stats LSMKeyStatistics, err error) {
 	var prevKey InternalKey
 	var rateLimitFunc func(key *InternalKey, val LazyValue) error
 	tb := tokenbucket.TokenBucket{}
@@ -2767,14 +2766,12 @@ func (d *DB) ScanStatistics(
 	if err != nil {
 		return LSMKeyStatistics{}, err
 	}
-	defer iter.close()
+	defer func() { err = errors.CombineErrors(err, iter.Close()) }()
 
 	err = scanInternalImpl(ctx, iter, scanInternalOpts)
-
 	if err != nil {
 		return LSMKeyStatistics{}, err
 	}
-
 	return stats, nil
 }
 

scan_internal.go

@@ -130,12 +130,13 @@ func (s *SharedSSTMeta) cloneFromFileMeta(f *manifest.TableMetadata) {
 // creator ID was set (as creator IDs are necessary to enable shared storage)
 // resulting in some lower level SSTs being on non-shared storage. Skip-shared
 // iteration is invalid in those cases.
-func (d *DB) ScanInternal(ctx context.Context, opts ScanInternalOptions) error {
-	iter, err := d.newInternalIter(ctx, snapshotIterOpts{} /* snapshot */, &opts)
+func (d *DB) ScanInternal(ctx context.Context, opts ScanInternalOptions) (err error) {
+	var iter *scanInternalIterator
+	iter, err = d.newInternalIter(ctx, snapshotIterOpts{} /* snapshot */, &opts)
 	if err != nil {
 		return err
 	}
-	defer iter.close()
+	defer func() { err = errors.CombineErrors(err, iter.Close()) }()
 	return scanInternalImpl(ctx, iter, &opts)
 }
 
@@ -1277,10 +1278,10 @@ func (i *scanInternalIterator) error() error {
 	return i.iter.Error()
 }
 
-// close closes this iterator, and releases any pooled objects.
-func (i *scanInternalIterator) close() {
-	_ = i.iter.Close()
-	_ = i.blobValueFetcher.Close()
+// Close closes this iterator, and releases any pooled objects.
+func (i *scanInternalIterator) Close() error {
+	err := i.iter.Close()
+	err = errors.CombineErrors(err, i.blobValueFetcher.Close())
 	if i.readState != nil {
 		i.readState.unref()
 	}
@@ -1311,6 +1312,7 @@ func (i *scanInternalIterator) close() {
 		iterAllocPool.Put(alloc)
 		i.alloc = nil
 	}
+	return err
 }
 
 func (i *scanInternalIterator) initializeBoundBufs(lower, upper []byte) {

snapshot.go

@@ -11,6 +11,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/manifest"
@@ -72,15 +73,16 @@ func (s *Snapshot) NewIterWithContext(ctx context.Context, o *IterOptions) (*Ite
 //
 // See comment on db.ScanInternal for the behaviour that can be expected of
 // point keys deleted by range dels and keys masked by range keys.
-func (s *Snapshot) ScanInternal(ctx context.Context, opts ScanInternalOptions) error {
+func (s *Snapshot) ScanInternal(ctx context.Context, opts ScanInternalOptions) (err error) {
 	if s.db == nil {
 		panic(ErrClosed)
 	}
-	iter, err := s.db.newInternalIter(ctx, snapshotIterOpts{seqNum: s.seqNum}, &opts)
+	var iter *scanInternalIterator
+	iter, err = s.db.newInternalIter(ctx, snapshotIterOpts{seqNum: s.seqNum}, &opts)
 	if err != nil {
 		return err
 	}
-	defer iter.close()
+	defer func() { err = errors.CombineErrors(err, iter.Close()) }()
 	return scanInternalImpl(ctx, iter, &opts)
 }
 
@@ -449,7 +451,7 @@ func (es *EventuallyFileOnlySnapshot) NewIterWithContext(
 // point keys deleted by range dels and keys masked by range keys.
 func (es *EventuallyFileOnlySnapshot) ScanInternal(
 	ctx context.Context, opts ScanInternalOptions,
-) error {
+) (err error) {
 	if es.db == nil {
 		panic(ErrClosed)
 	}
@@ -467,10 +469,11 @@ func (es *EventuallyFileOnlySnapshot) ScanInternal(
 		}
 	}
 	es.mu.Unlock()
-	iter, err := es.db.newInternalIter(ctx, sOpts, &opts)
+	var iter *scanInternalIterator
+	iter, err = es.db.newInternalIter(ctx, sOpts, &opts)
 	if err != nil {
 		return err
 	}
-	defer iter.close()
+	defer func() { err = errors.CombineErrors(err, iter.Close()) }()
 	return scanInternalImpl(ctx, iter, &opts)
 }