db: add support for reading blob values

Initialize iterators and compactions with a blob.ValueFetcher that's propagated
through to sstable iterators. When an sstbale iterator encounters a blob value
handle, it constructs an internal value with the blob.ValueFetcher configured
as the ValueFetcher.

Close #4394.
Informs #112.

Author: jbowens

compaction.go

@@ -29,6 +29,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 )
@@ -216,6 +217,9 @@ type compaction struct {
 	// blob files. This consumes more write bandwidth because all values are
 	// rewritten. However it restores locality.
 	getValueSeparation func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation
+	// valueFetcher is used to fetch values from blob files. It's propagated
+	// down the iterator tree through the internal iterator options.
+	valueFetcher blob.ValueFetcher
 
 	// startLevel is the level that is being compacted. Inputs from startLevel
 	// and outputLevel will be merged to produce a set of outputLevel files.
@@ -3085,16 +3089,19 @@ func (d *DB) compactAndWrite(
 	// translate to 3 MiB per compaction.
 	c.bufferPool.Init(12)
 	defer c.bufferPool.Release()
+	env := block.ReadEnv{
+		BufferPool: &c.bufferPool,
+		Stats:      &c.stats,
+		IterStats: d.fileCache.SSTStatsCollector().Accumulator(
+			uint64(uintptr(unsafe.Pointer(c))),
+			categoryCompaction,
+		),
+	}
+	c.valueFetcher.Init(d.fileCache, env)
 	iiopts := internalIterOpts{
-		compaction: true,
-		readEnv: block.ReadEnv{
-			BufferPool: &c.bufferPool,
-			Stats:      &c.stats,
-			IterStats: d.fileCache.SSTStatsCollector().Accumulator(
-				uint64(uintptr(unsafe.Pointer(c))),
-				categoryCompaction,
-			),
-		},
+		compaction:       true,
+		readEnv:          env,
+		blobValueFetcher: &c.valueFetcher,
 	}
 
 	pointIter, rangeDelIter, rangeKeyIter, err := c.newInputIters(d.newIters, d.tableNewRangeKeyIter, iiopts)

compaction_test.go

@@ -610,7 +610,7 @@ func TestCompaction(t *testing.T) {
 					return "", "", errors.WithStack(err)
 				}
 				defer r.Close()
-				iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+				iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 				if err != nil {
 					return "", "", errors.WithStack(err)
 				}

db.go

@@ -1349,6 +1349,9 @@ type internalIterOpts struct {
 	compaction         bool
 	readEnv            block.ReadEnv
 	boundLimitedFilter sstable.BoundLimitedBlockPropertyFilter
+	// blobValueFetcher is the base.ValueFetcher to use when constructing
+	// internal values to represent values stored externally in blob files.
+	blobValueFetcher base.ValueFetcher
 }
 
 func finishInitializingInternalIter(
@@ -1401,18 +1404,21 @@ func (i *Iterator) constructPointIter(
 		// Already have one.
 		return
 	}
+	readEnv := block.ReadEnv{
+		Stats: &i.stats.InternalStats,
+		// If the file cache has a sstable stats collector, ask it for an
+		// accumulator for this iterator's configured category and QoS. All SSTable
+		// iterators created by this Iterator will accumulate their stats to it as
+		// they Close during iteration.
+		IterStats: i.fc.SSTStatsCollector().Accumulator(
+			uint64(uintptr(unsafe.Pointer(i))),
+			i.opts.Category,
+		),
+	}
+	i.blobValueFetcher.Init(i.fc, readEnv)
 	internalOpts := internalIterOpts{
-		readEnv: block.ReadEnv{
-			Stats: &i.stats.InternalStats,
-			// If the file cache has a sstable stats collector, ask it for an
-			// accumulator for this iterator's configured category and QoS. All SSTable
-			// iterators created by this Iterator will accumulate their stats to it as
-			// they Close during iteration.
-			IterStats: i.fc.SSTStatsCollector().Accumulator(
-				uint64(uintptr(unsafe.Pointer(i))),
-				i.opts.Category,
-			),
-		},
+		readEnv:          readEnv,
+		blobValueFetcher: &i.blobValueFetcher,
 	}
 	if i.opts.RangeKeyMasking.Filter != nil {
 		internalOpts.boundLimitedFilter = &i.rangeKeyMasking

file_cache.go

@@ -677,7 +677,11 @@ func (h *fileCacheHandle) newPointIter(
 		internalOpts.readEnv.IterStats = handle.SSTStatsCollector().Accumulator(uint64(uintptr(unsafe.Pointer(r))), opts.Category)
 	}
 	if internalOpts.compaction {
-		iter, err = cr.NewCompactionIter(transforms, internalOpts.readEnv, &v.readerProvider)
+		iter, err = cr.NewCompactionIter(transforms, internalOpts.readEnv,
+			&v.readerProvider, sstable.TableBlobContext{
+				ValueFetcher: internalOpts.blobValueFetcher,
+				References:   file.BlobReferences,
+			})
 	} else {
 		iter, err = cr.NewPointIter(ctx, sstable.IterOptions{
 			Lower:                opts.GetLowerBound(),
@@ -687,6 +691,10 @@ func (h *fileCacheHandle) newPointIter(
 			Filterer:             filterer,
 			Env:                  internalOpts.readEnv,
 			ReaderProvider:       &v.readerProvider,
+			BlobContext: sstable.TableBlobContext{
+				ValueFetcher: internalOpts.blobValueFetcher,
+				References:   file.BlobReferences,
+			},
 		})
 	}
 	if err != nil {

ingest.go

@@ -360,7 +360,7 @@ func ingestLoad1(
 	maybeSetStatsFromProperties(meta.PhysicalMeta(), &r.Properties)
 
 	{
-		iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+		iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 		if err != nil {
 			return nil, keyspan.Span{}, err
 		}

internal/base/iterator.go

@@ -416,6 +416,9 @@ type InternalIteratorStats struct {
 		// ValueBytesFetched is the total byte length of the values (in value
 		// blocks) that were retrieved.
 		ValueBytesFetched uint64
+
+		// TODO(jackson): Add stats for distinguishing between value-block
+		// values and blob values.
 	}
 }
 

internal/base/lazy_value.go

@@ -4,7 +4,11 @@
 
 package base
 
-import "context"
+import (
+	"context"
+
+	"github.com/cockroachdb/errors"
+)
 
 // A value can have user-defined attributes that are a function of the value
 // byte slice. For now, we only support "short attributes", which can be
@@ -278,3 +282,26 @@ func (lv *LazyValue) Clone(buf []byte, fetcher *LazyFetcher) (LazyValue, []byte)
 	lvCopy.ValueOrHandle = buf[bufLen : bufLen+vLen]
 	return lvCopy, buf
 }
+
+// NoBlobFetches is a ValueFetcher that returns an error. It's intended to be
+// used in situations where sstables should not encode a blob value, or the
+// caller should not fetch the handle's value.
+var NoBlobFetches = &errValueFetcher{
+	Err: errors.AssertionFailedf("unexpected blob value"),
+}
+
+// errValueFetcher is a ValueFetcher that returns an error.
+type errValueFetcher struct {
+	Err error
+}
+
+// Assert that *errValueFetcher implements base.ValueFetcher.
+var _ ValueFetcher = (*errValueFetcher)(nil)
+
+// Fetch implements base.ValueFetcher.
+func (e *errValueFetcher) Fetch(
+	_ context.Context, _ []byte, blobFileNum DiskFileNum, valLen uint32, _ []byte,
+) (val []byte, callerOwned bool, err error) {
+	err = errors.Wrapf(e.Err, "fetching %d-byte value from %s", valLen, blobFileNum)
+	return nil, false, err
+}

internal/blobtest/handles.go

@@ -26,7 +26,8 @@ import (
 // human-readable string describing a blob handle, synthesizing unspecified
 // fields, and tracking the blob handle to support future fetches.
 type Values struct {
-	mostRecentHandle blob.Handle
+	mostRecentFileNum base.DiskFileNum
+	mostRecentHandles map[base.DiskFileNum]blob.Handle
 	// trackedHandles maps from a blob handle to its value. The value may be nil
 	// if the value was not specified (in which case Fetch will
 	// deterministically derive a random value from the handle itself.)
@@ -108,6 +109,7 @@ func (bv *Values) IsBlobHandle(input string) bool {
 func (bv *Values) Parse(input string) (h blob.Handle, err error) {
 	if bv.trackedHandles == nil {
 		bv.trackedHandles = make(map[blob.Handle]string)
+		bv.mostRecentHandles = make(map[base.DiskFileNum]blob.Handle)
 	}
 
 	defer func() {
@@ -156,13 +158,13 @@ func (bv *Values) Parse(input string) (h blob.Handle, err error) {
 	}
 
 	if !fileNumSet {
-		h.FileNum = max(bv.mostRecentHandle.FileNum, 1)
+		h.FileNum = bv.mostRecentFileNum
 	}
 	if !blockNumSet {
-		h.BlockNum = bv.mostRecentHandle.BlockNum
+		h.BlockNum = bv.mostRecentHandles[h.FileNum].BlockNum
 	}
 	if !offsetSet {
-		h.OffsetInBlock = bv.mostRecentHandle.OffsetInBlock + bv.mostRecentHandle.ValueLen
+		h.OffsetInBlock = bv.mostRecentHandles[h.FileNum].OffsetInBlock + bv.mostRecentHandles[h.FileNum].ValueLen
 	}
 	if !valueLenSet {
 		if len(value) > 0 {
@@ -171,7 +173,8 @@ func (bv *Values) Parse(input string) (h blob.Handle, err error) {
 			h.ValueLen = 12
 		}
 	}
-	bv.mostRecentHandle = h
+	bv.mostRecentFileNum = h.FileNum
+	bv.mostRecentHandles[h.FileNum] = h
 	bv.trackedHandles[h] = value
 	return h, nil
 }
@@ -228,7 +231,10 @@ func WriteFiles(
 			return err
 		}
 		writer := blob.NewFileWriter(fileNum, writable, writerOpts)
-		for _, handle := range handles {
+		for i, handle := range handles {
+			if i > 0 && handles[i-1].BlockNum != handle.BlockNum {
+				writer.FlushForTesting()
+			}
 			if value, ok := bv.trackedHandles[handle]; ok {
 				writer.AddValue([]byte(value))
 			} else {

internal/compact/spans_test.go

@@ -135,7 +135,7 @@ func TestSplitAndEncodeSpan(t *testing.T) {
 			_, rangeDels, rangeKeys, err := sstable.ReadAll(obj, sstable.ReaderOptions{
 				Comparer:   wo.Comparer,
 				KeySchemas: sstable.MakeKeySchemas(wo.KeySchema),
-			})
+			}, base.NoBlobFetches)
 			require.NoError(t, err)
 			require.LessOrEqual(t, len(rangeDels)+len(rangeKeys), 1)
 			s := "."

internal/manifest/blob_metadata.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/strparse"
+	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/redact"
 )
@@ -224,6 +225,9 @@ type BlobReferenceDepth int
 // persisted to the manifest.
 type BlobReferences []BlobReference
 
+// Assert that BlobReferences implements sstable.BlobReferences.
+var _ sstable.BlobReferences = BlobReferences{}
+
 // FileNumByID returns the FileNum for the identified BlobReference.
 func (br BlobReferences) FileNumByID(i blob.ReferenceID) base.DiskFileNum {
 	return br[i].FileNum