internal/compact: preserve blob references in Iter

Adapt the compaction iterator to preserve blob file references by cloning the
LazyValues. Compactions that do not rewrite blob files will avoid ever
retrieving the corresponding values.

Additionally, fix LazyValue.Clone to copy the BlobFileNum of the LazyFetcher.

Informs #112.

Author: jbowens

internal/base/lazy_value.go

@@ -216,7 +216,7 @@ func (lv *LazyValue) fetchValue(
 	if !f.fetched {
 		f.fetched = true
 		f.value, f.callerOwned, f.err = f.Fetcher.Fetch(ctx,
-			lv.ValueOrHandle, lv.Fetcher.BlobFileNum, lv.Fetcher.Attribute.ValueLen, buf)
+			lv.ValueOrHandle, f.BlobFileNum, f.Attribute.ValueLen, buf)
 	}
 	return f.value, f.callerOwned, f.err
 }
@@ -262,8 +262,9 @@ func (lv *LazyValue) Clone(buf []byte, fetcher *LazyFetcher) (LazyValue, []byte)
 	var lvCopy LazyValue
 	if lv.Fetcher != nil {
 		*fetcher = LazyFetcher{
-			Fetcher:   lv.Fetcher.Fetcher,
-			Attribute: lv.Fetcher.Attribute,
+			Fetcher:     lv.Fetcher.Fetcher,
+			Attribute:   lv.Fetcher.Attribute,
+			BlobFileNum: lv.Fetcher.BlobFileNum,
 			// Not copying anything that has been extracted.
 		}
 		lvCopy.Fetcher = fetcher

internal/base/lazy_value_test.go

@@ -58,9 +58,11 @@ func TestLazyValue(t *testing.T) {
 						numCalls++
 						require.Equal(t, []byte("foo-handle"), handle)
 						require.Equal(t, uint32(3), valLen)
+						require.Equal(t, DiskFileNum(90), blobFileNum)
 						return fooBytes1, callerOwned, nil
 					}),
-				Attribute: AttributeAndLen{ValueLen: 3, ShortAttribute: 7},
+				Attribute:   AttributeAndLen{ValueLen: 3, ShortAttribute: 7},
+				BlobFileNum: 90,
 			},
 		}
 		require.Equal(t, []byte("foo"), getValue(fooLV3, callerOwned))

internal/base/value.go

@@ -27,6 +27,13 @@ func MakeInPlaceValue(val []byte) InternalValue {
 	return InternalValue{lazyValue: LazyValue{ValueOrHandle: val}}
 }
 
+// IsBlobValueHandle returns true iff the value is a blob value handle, pointing
+// to a value stored externally in a blob file.
+func (v *InternalValue) IsBlobValueHandle() bool {
+	f := v.lazyValue.Fetcher
+	return f != nil && f.BlobFileNum > 0
+}
+
 // IsInPlaceValue returns true iff the value was stored in-place and does not
 // need to be fetched externally.
 func (v *InternalValue) IsInPlaceValue() bool {

internal/compact/iterator.go

@@ -192,7 +192,9 @@ type Iter struct {
 	// Temporary buffer used for storing the previous value, which may be an
 	// unsafe, i.iter-owned slice that could be altered when the iterator is
 	// advanced.
-	valueBuf         []byte
+	valueBuf []byte
+	// valueFetcher is used by saveValue when Cloning InternalValues.
+	valueFetcher     base.LazyFetcher
 	iterKV           *base.InternalKV
 	iterStripeChange stripeChangeType
 	// skip indicates whether the remaining entries in the current snapshot
@@ -878,6 +880,9 @@ func (i *Iter) mergeNext(valueMerger base.ValueMerger) {
 			// value and return. We change the kind of the resulting key to a
 			// Set so that it shadows keys in lower levels. That is:
 			// MERGE + (SET*) -> SET.
+			//
+			// Because we must merge the value, we must retrieve it regardless
+			// of whether the value is a blob reference.
 			var v []byte
 			var callerOwned bool
 			v, callerOwned, i.err = i.iterKV.Value(i.valueBuf[:0])
@@ -1289,11 +1294,20 @@ func (i *Iter) saveKey() {
 //
 // If the value is in-place, this copies it into i.valueBuf. If the value is in
 // a value block, it retrieves the value from the block (possibly storing the
-// result into i.valueBuf).
+// result into i.valueBuf). If the value is stored in an external blob file, the
+// value is cloned (InternalValue.Clone) without retrieving it from the external
+// file.
+//
+// Note that because saveValue uses i.valueBuf and i.valueFetcher to avoid
+// allocations, values saved by saveValue are only valid until the next call to
+// saveValue.
 func (i *Iter) saveValue() {
-	// TODO(jackson): With the introduction of values stored in separate
-	// physical blob files, this should begin to Clone LazyValues that are
-	// stored in blob reference files when non-rewriting blob files.
+	// Clone blob value handles to defer the retrieval of the value.
+	if i.iterKV.V.IsBlobValueHandle() {
+		i.kv.V, i.valueBuf = i.iterKV.V.Clone(i.valueBuf, &i.valueFetcher)
+		return
+	}
+
 	v, callerOwned, err := i.iterKV.Value(i.valueBuf[:0])
 	if err != nil {
 		i.err = err

internal/compact/iterator_test.go

@@ -6,18 +6,21 @@ package compact
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"io"
 	"slices"
 	"strconv"
 	"strings"
 	"testing"
+	"unicode"
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/rangekey"
+	"github.com/cockroachdb/pebble/sstable/valblk"
 	"github.com/stretchr/testify/require"
 )
 
@@ -53,7 +56,6 @@ func TestCompactionIter(t *testing.T) {
 	var snapshots Snapshots
 	var elideTombstones bool
 	var allowZeroSeqnum bool
-
 	var ineffectualSingleDeleteKeys []string
 	var invariantViolationSingleDeleteKeys []string
 	resetSingleDelStats := func() {
@@ -133,16 +135,18 @@ func TestCompactionIter(t *testing.T) {
 						continue
 					}
 
-					var value []byte
+					var iv base.InternalValue
 					if strings.HasPrefix(key[j+1:], "varint(") {
 						valueStr := strings.TrimSuffix(strings.TrimPrefix(key[j+1:], "varint("), ")")
 						v, err := strconv.ParseUint(valueStr, 10, 64)
 						require.NoError(t, err)
-						value = binary.AppendUvarint([]byte(nil), v)
+						iv = base.MakeInPlaceValue(binary.AppendUvarint([]byte(nil), v))
+					} else if strings.HasPrefix(key[j+1:], "blobref(") {
+						iv = decodeBlobReference(t, key[j+1:])
 					} else {
-						value = []byte(key[j+1:])
+						iv = base.MakeInPlaceValue([]byte(key[j+1:]))
 					}
-					kvs = append(kvs, base.MakeInternalKV(ik, value))
+					kvs = append(kvs, base.InternalKV{K: ik, V: iv})
 				}
 				rangeDelFragmenter.Finish()
 				return ""
@@ -202,9 +206,15 @@ func TestCompactionIter(t *testing.T) {
 					}
 					var value []byte
 					if kv != nil {
-						var err error
-						value, _, err = kv.Value(nil)
-						require.NoError(t, err)
+						if kv.V.IsBlobValueHandle() {
+							lv := kv.V.LazyValue()
+							value = []byte(fmt.Sprintf("<blobref(%s, encodedHandle=%x, valLen=%d)>",
+								lv.Fetcher.BlobFileNum, lv.ValueOrHandle, lv.Fetcher.Attribute.ValueLen))
+						} else {
+							var err error
+							value, _, err = kv.Value(nil)
+							require.NoError(t, err)
+						}
 					}
 
 					if kv != nil {
@@ -267,6 +277,60 @@ func TestCompactionIter(t *testing.T) {
 	runTest(t, "testdata/iter_delete_sized")
 }
 
+// mockBlobValueFetcher is a dummy ValueFetcher implementation which produces
+// string values referencing the blobref.
+type mockBlobValueFetcher struct{}
+
+var _ base.ValueFetcher = mockBlobValueFetcher{}
+
+func (mockBlobValueFetcher) Fetch(
+	ctx context.Context, handle []byte, fileNum base.DiskFileNum, valLen uint32, buf []byte,
+) (val []byte, callerOwned bool, err error) {
+	s := fmt.Sprintf("<fetched value from blobref(%s, encodedHandle=%x, valLen=%d)>",
+		fileNum, handle, valLen)
+	return []byte(s), false, nil
+}
+
+// decodeBlobReference decodes a blob reference from a debug string. It expects
+// a value of the form: blobref(<filenum>, blk<blocknum>, <offset>, <valLen>).
+// For example: blobref(000124, blk255, 10, 9235)
+func decodeBlobReference(t testing.TB, ref string) base.InternalValue {
+	fields := strings.FieldsFunc(strings.TrimSuffix(strings.TrimPrefix(ref, "blobref("), ")"),
+		func(r rune) bool { return r == ',' || unicode.IsSpace(r) })
+	require.Equal(t, 4, len(fields))
+	fileNum, err := strconv.ParseUint(fields[0], 10, 64)
+	require.NoError(t, err)
+	blockNum, err := strconv.ParseUint(strings.TrimPrefix(fields[1], "blk"), 10, 32)
+	require.NoError(t, err)
+	off, err := strconv.ParseUint(fields[2], 10, 32)
+	require.NoError(t, err)
+	valLen, err := strconv.ParseUint(fields[3], 10, 32)
+	require.NoError(t, err)
+
+	// TODO(jackson): Support short (and long, when introduced) attributes.
+	return base.MakeLazyValue(base.LazyValue{
+		ValueOrHandle: encodeRemainingHandle(uint32(blockNum), uint32(off)),
+		Fetcher: &base.LazyFetcher{
+			Fetcher:     mockBlobValueFetcher{},
+			BlobFileNum: base.DiskFileNum(fileNum),
+			Attribute: base.AttributeAndLen{
+				ValueLen: uint32(valLen),
+			},
+		},
+	})
+}
+
+func encodeRemainingHandle(blockNum uint32, offsetInBlock uint32) []byte {
+	// TODO(jackson): Pull this into a common helper.
+	dst := make([]byte, valblk.HandleMaxLen)
+	n := valblk.EncodeHandle(dst, valblk.Handle{
+		ValueLen:      0,
+		BlockNum:      blockNum,
+		OffsetInBlock: offsetInBlock,
+	})
+	return dst[1:n]
+}
+
 // makeInputIters creates the iterators necessthat can be used to create a compaction
 // Iter.
 func makeInputIters(

internal/compact/testdata/iter

@@ -1124,3 +1124,44 @@ next
 ----
 a#3,SET:val
 .
+
+# Define an input sequence of simple blob references, all pointing into the same
+# blob file block.
+
+define
+a.SET.9:blobref(000294, blk2, 10, 20)
+b.SET.3:blobref(000294, blk2, 30, 5)
+c.SETWITHDEL.8:blobref(000294, blk2, 35, 100)
+d.SET.2:blobref(000294, blk2, 135, 4)
+----
+
+# An iterator should preserve blob references.
+
+iter
+first
+next
+next
+next
+----
+a#9,SET:<blobref(000294, encodedHandle=020a, valLen=20)>
+b#3,SET:<blobref(000294, encodedHandle=021e, valLen=5)>
+c#8,SETWITHDEL:<blobref(000294, encodedHandle=0223, valLen=100)>
+d#2,SET:<blobref(000294, encodedHandle=028701, valLen=4)>
+
+# The iterator may need to fetch a blob value if it's an operand to the merge
+# operator.
+
+define
+a.SET.3:blobref(000294, blk2, 10, 20)
+b.MERGE.9:mergekeyvalue
+b.SETWITHDEL.8:blobref(000294, blk2, 35, 100)
+----
+
+iter
+first
+next
+next
+----
+a#3,SET:<blobref(000294, encodedHandle=020a, valLen=20)>
+b#9,SET:<fetched value from blobref(000294, encodedHandle=0223, valLen=100)>mergekeyvalue[base]
+.