db: support blob values in the level checker

Update the level checker to initialize a blob value fetcher and use it.

Author: jbowens

level_checker.go

@@ -17,6 +17,9 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/blob"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 // This file implements DB.CheckLevels() which checks that every entry in the
@@ -353,6 +356,10 @@ type checkConfig struct {
 	stats     *CheckLevelsStats
 	merge     Merge
 	formatKey base.FormatKey
+	readEnv   block.ReadEnv
+	// blobValueFetcher is the ValueFetcher to use when retrieving values stored
+	// externally in blob files.
+	blobValueFetcher blob.ValueFetcher
 }
 
 // cmp is shorthand for comparer.Compare.
@@ -538,11 +545,21 @@ func (d *DB) CheckLevels(stats *CheckLevelsStats) error {
 		stats:     stats,
 		merge:     d.merge,
 		formatKey: d.opts.Comparer.FormatKey,
+		readEnv:   block.ReadEnv{
+			// TODO(jackson): Add categorized stats.
+		},
 	}
+	checkConfig.blobValueFetcher.Init(d.fileCache, checkConfig.readEnv)
+	defer func() { _ = checkConfig.blobValueFetcher.Close() }()
 	return checkLevelsInternal(checkConfig)
 }
 
 func checkLevelsInternal(c *checkConfig) (err error) {
+	internalOpts := internalIterOpts{
+		readEnv:          sstable.ReadEnv{Block: c.readEnv},
+		blobValueFetcher: &c.blobValueFetcher,
+	}
+
 	// Phase 1: Use a simpleMergingIter to step through all the points and ensure
 	// that points with the same user key at different levels are not inverted
 	// wrt sequence numbers and the same holds for tombstones that cover points.
@@ -600,7 +617,7 @@ func checkLevelsInternal(c *checkConfig) (err error) {
 		iterOpts := IterOptions{logger: c.logger}
 		li := &levelIter{}
 		li.init(context.Background(), iterOpts, c.comparer, c.newIters, manifestIter,
-			manifest.L0Sublevel(sublevel), internalIterOpts{})
+			manifest.L0Sublevel(sublevel), internalOpts)
 		li.initRangeDel(&mlevelAlloc[0])
 		mlevelAlloc[0].iter = li
 		mlevelAlloc = mlevelAlloc[1:]
@@ -613,7 +630,7 @@ func checkLevelsInternal(c *checkConfig) (err error) {
 		iterOpts := IterOptions{logger: c.logger}
 		li := &levelIter{}
 		li.init(context.Background(), iterOpts, c.comparer, c.newIters,
-			current.Levels[level].Iter(), manifest.Level(level), internalIterOpts{})
+			current.Levels[level].Iter(), manifest.Level(level), internalOpts)
 		li.initRangeDel(&mlevelAlloc[0])
 		mlevelAlloc[0].iter = li
 		mlevelAlloc = mlevelAlloc[1:]