db: separate mvcc garbage into blob files

This patch separates SET keys that have already appeared in the
sstable into a blob file.

Informs: #4424

Author: annrpom

blob_rewrite_test.go

@@ -120,7 +120,7 @@ func TestBlobRewrite(t *testing.T) {
 					} else {
 						ikv.V = base.MakeInPlaceValue([]byte(parts[1]))
 					}
-					require.NoError(t, vs.Add(tw, &ikv, false /* forceObsolete */))
+					require.NoError(t, vs.Add(tw, &ikv, false /* forceObsolete */, func() bool { return false } /* isLikeyMVCCGarbage */))
 				}
 				return buf.String()
 			case "close-output":

compaction_test.go

@@ -908,7 +908,7 @@ func TestCompaction(t *testing.T) {
 			fmt.Fprintln(&compactionLog, info.String())
 		},
 	}
-	reset := func(minVersion, maxVersion FormatMajorVersion) {
+	reset := func(minVersion, maxVersion FormatMajorVersion, cmp *Comparer) {
 		compactionLog.Reset()
 		if d != nil {
 			require.NoError(t, closeAllSnapshots(d))
@@ -924,6 +924,7 @@ func TestCompaction(t *testing.T) {
 			EventListener:               compactionLogEventListener,
 			FormatMajorVersion:          randVersion(minVersion, maxVersion),
 			Logger:                      testutils.Logger{T: t},
+			Comparer:                    cmp,
 		}
 		opts.WithFSDefaults()
 		opts.Experimental.CompactionScheduler = NewConcurrencyLimitSchedulerWithNoPeriodicGrantingForTest()
@@ -971,13 +972,13 @@ func TestCompaction(t *testing.T) {
 		d.mu.compact.compactingCount--
 	}
 
-	runTest := func(t *testing.T, testData string, minVersion, maxVersion FormatMajorVersion, verbose bool) {
-		reset(minVersion, maxVersion)
+	runTest := func(t *testing.T, testData string, minVersion, maxVersion FormatMajorVersion, verbose bool, cmp *Comparer) {
+		reset(minVersion, maxVersion, cmp)
 		var ongoingCompaction *tableCompaction
 		datadriven.RunTest(t, testData, func(t *testing.T, td *datadriven.TestData) string {
 			switch td.Cmd {
 			case "reset":
-				reset(minVersion, maxVersion)
+				reset(minVersion, maxVersion, cmp)
 				return ""
 
 			case "batch":
@@ -1028,6 +1029,9 @@ func TestCompaction(t *testing.T) {
 					DisableAutomaticCompactions: true,
 					Logger:                      testutils.Logger{T: t},
 				}
+				if cmp != nil {
+					opts.Comparer = cmp
+				}
 				opts.WithFSDefaults()
 				opts.Experimental.CompactionScheduler = NewConcurrencyLimitSchedulerWithNoPeriodicGrantingForTest()
 				if d != nil {
@@ -1453,6 +1457,7 @@ func TestCompaction(t *testing.T) {
 		minVersion FormatMajorVersion // inclusive, FormatMinSupported if unspecified.
 		maxVersion FormatMajorVersion // inclusive, internalFormatNewest if unspecified.
 		verbose    bool
+		cmp        *Comparer
 	}
 	testConfigs := map[string]testConfig{
 		"singledel_set_with_del": {},
@@ -1483,6 +1488,12 @@ func TestCompaction(t *testing.T) {
 			maxVersion: FormatValueSeparation,
 			verbose:    true,
 		},
+		"mvcc_garbage_blob": {
+			minVersion: FormatValueSeparation,
+			maxVersion: FormatValueSeparation,
+			verbose:    true,
+			cmp:        testkeys.Comparer,
+		},
 		"score_compaction_picked_before_manual": {
 			// Run at a specific version, so that a single sstable format is used,
 			// since the test prints the compaction log which includes file sizes.
@@ -1513,7 +1524,7 @@ func TestCompaction(t *testing.T) {
 		if maxVersion == 0 {
 			maxVersion = internalFormatNewest
 		}
-		runTest(t, path, minVersion, maxVersion, tc.verbose)
+		runTest(t, path, minVersion, maxVersion, tc.verbose, tc.cmp)
 	})
 }
 
@@ -1592,14 +1603,14 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 			// Collection of table stats can trigger compactions. As we want full
 			// control over when compactions are run, disable stats by default.
 			DisableTableStats: true,
-			//EventListener: &EventListener{
+			// EventListener: &EventListener{
 			//	CompactionEnd: func(info CompactionInfo) {
 			//		if compactInfo != nil {
 			//			return
 			//		}
 			//		compactInfo = &info
 			//	},
-			//},
+			// },
 			EventListener:      &el,
 			FormatMajorVersion: internalFormatNewest,
 			Logger:             testutils.Logger{T: t},

internal/compact/run.go

@@ -133,7 +133,7 @@ type ValueSeparation interface {
 	EstimatedReferenceSize() uint64
 	// Add adds the provided key-value pair to the provided sstable writer,
 	// possibly separating the value into a blob file.
-	Add(tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool) error
+	Add(tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, isLikelyMVCCGarbage func() bool) error
 	// FinishOutput is called when a compaction is finishing an output sstable.
 	// It returns the table's blob references, which will be added to the
 	// table's TableMetadata, and stats and metadata describing a newly
@@ -292,8 +292,12 @@ func (r *Runner) writeKeysToTable(
 	equalPrev := func(k []byte) bool {
 		return tw.ComparePrev(k) == 0
 	}
+	prefixEqual := func(k []byte) bool {
+		return tw.IsPrefixEqualPrev(k)
+	}
 	var pinnedKeySize, pinnedValueSize, pinnedCount uint64
 	var iteratedKeys uint64
+	var prevKeyKind base.InternalKeyKind
 	kv := r.kv
 	for ; kv != nil; kv = r.iter.Next() {
 		iteratedKeys++
@@ -331,10 +335,17 @@ func (r *Runner) writeKeysToTable(
 		}
 
 		valueLen := kv.V.Len()
+		isLikelyMVCCGarbage := func() bool {
+			return sstable.IsLikelyMVCCGarbage(kv.K.UserKey, prevKeyKind, kv.K.Kind(), valueLen, prefixEqual)
+		}
+		prevKeyKind = kv.K.Kind()
 		// Add the value to the sstable, possibly separating its value into a
 		// blob file. The ValueSeparation implementation is responsible for
 		// writing the KV to the sstable.
-		if err := valueSeparation.Add(tw, kv, r.iter.ForceObsoleteDueToRangeDel()); err != nil {
+		// If the key might be garbage (all requirements of
+		// sstable.IsLikelyMVCCGarbage are met), we eagerly separate the value
+		// into a blob file.
+		if err := valueSeparation.Add(tw, kv, r.iter.ForceObsoleteDueToRangeDel(), isLikelyMVCCGarbage); err != nil {
 			return nil, err
 		}
 		if r.iter.SnapshotPinned() {
@@ -497,7 +508,7 @@ func (NeverSeparateValues) EstimatedReferenceSize() uint64 { return 0 }
 
 // Add implements the ValueSeparation interface.
 func (NeverSeparateValues) Add(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool,
+	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ func() bool,
 ) error {
 	v, _, err := kv.Value(nil)
 	if err != nil {

sstable/colblk_writer.go

@@ -256,6 +256,19 @@ func (w *RawColumnWriter) ComparePrev(k []byte) int {
 	return int(w.dataBlock.KeyWriter.ComparePrev(k).UserKeyComparison)
 }
 
+// IsPrefixEqualPrev compares the provided user key's prefix to the key
+// prefix of the last point key written to the writer.
+//
+// If no key has been written yet, IsPrefixEqualPrev returns false.
+//
+// Must not be called after Writer is closed.
+func (w *RawColumnWriter) IsPrefixEqualPrev(k []byte) bool {
+	if w == nil || w.dataBlock.Rows() == 0 {
+		return false
+	}
+	return w.dataBlock.KeyWriter.ComparePrev(k).PrefixEqual()
+}
+
 // SetSnapshotPinnedProperties sets the properties for pinned keys. Should only
 // be used internally by Pebble.
 func (w *RawColumnWriter) SetSnapshotPinnedProperties(
@@ -621,30 +634,18 @@ func (w *RawColumnWriter) evaluatePoint(
 			key.Pretty(w.comparer.FormatKey))
 	}
 
+	prefixEqual := func(k []byte) bool {
+		return w.IsPrefixEqualPrev(k)
+	}
 	// We might want to write this key's value to a value block if it has the
 	// same prefix.
 	//
 	// We require:
 	//  . Value blocks to be enabled.
-	//  . The current key to have the same prefix as the previous key.
-	//  . The previous key to be a SET.
-	//  . The current key to be a SET.
-	//  . If there are bounds requiring some keys' values to be in-place, the
-	//    key must not fall within those bounds.
-	//  . The value to be sufficiently large. (Currently we simply require a
-	//    non-zero length, so all non-empty values are eligible for storage
-	//    out-of-band in a value block.)
-	//
-	// Use of 0 here is somewhat arbitrary. Given the minimum 3 byte encoding of
-	// valueHandle, this should be > 3. But tiny values are common in test and
-	// unlikely in production, so we use 0 here for better test coverage.
-	const tinyValueThreshold = 0
+	//  . IsLikelyMVCCGarbage to be true; see comment for MVCC garbage criteria.
 	useValueBlock := !w.opts.DisableValueBlocks &&
-		eval.kcmp.PrefixEqual() &&
-		prevKeyKind == InternalKeyKindSet &&
-		keyKind == InternalKeyKindSet &&
-		valueLen > tinyValueThreshold &&
-		w.valueBlock != nil
+		w.valueBlock != nil &&
+		IsLikelyMVCCGarbage(key.UserKey, prevKeyKind, keyKind, valueLen, prefixEqual)
 	if !useValueBlock {
 		return eval, nil
 	}
@@ -1270,3 +1271,31 @@ func (w *RawColumnWriter) copyProperties(props Properties) {
 	w.props.IndexSize = 0
 	w.props.IndexType = 0
 }
+
+// IsLikelyMVCCGarbage determines whether the given user key is likely MVCC
+// garbage.
+//
+// We require:
+//
+//	. The previous key to be a SET.
+//	. The current key to be a SET.
+//	. The value to be sufficiently large. (Currently we simply require a
+//	  non-zero length, so all non-empty values are eligible for storage
+//	  out-of-band in a value block.)
+//	. The current key to have the same prefix as the previous key.
+//
+// Use of 0 here is somewhat arbitrary. Given the minimum 3 byte encoding of
+// valueHandle, this should be > 3. But tiny values are common in test and
+// unlikely in production, so we use 0 here for better test coverage.
+func IsLikelyMVCCGarbage(
+	k []byte,
+	prevKeyKind, keyKind base.InternalKeyKind,
+	valueLen int,
+	prefixEqual func(k []byte) bool,
+) bool {
+	const tinyValueThreshold = 0
+	return prevKeyKind == InternalKeyKindSet &&
+		keyKind == InternalKeyKindSet &&
+		valueLen > tinyValueThreshold &&
+		prefixEqual(k)
+}

sstable/rowblk_writer.go

@@ -458,7 +458,7 @@ type dataBlockBuf struct {
 }
 
 func (d *dataBlockBuf) clear() {
-	//d.blockBuf.clear()
+	// d.blockBuf.clear()
 	d.dataBlock.Reset()
 
 	d.uncompressed = nil
@@ -1395,6 +1395,19 @@ func (w *RawRowWriter) ComparePrev(k []byte) int {
 	return w.compare(k, w.dataBlockBuf.dataBlock.CurUserKey())
 }
 
+// IsPrefixEqualPrev compares the provided user key's prefix to the key
+// prefix of the last point key written to the writer.
+//
+// If no key has been written yet, IsPrefixEqualPrev returns false.
+//
+// Must not be called after Writer is closed.
+func (w *RawRowWriter) IsPrefixEqualPrev(k []byte) bool {
+	if w == nil || w.dataBlockBuf.dataBlock.EntryCount() == 0 {
+		return false
+	}
+	return bytes.Equal(w.split.Prefix(k), w.split.Prefix(w.dataBlockBuf.dataBlock.CurUserKey()))
+}
+
 // EncodeSpan encodes the keys in the given span. The span can contain either
 // only RANGEDEL keys or only range keys.
 //

sstable/writer.go

@@ -337,9 +337,9 @@ type RawWriter interface {
 	// EstimatedSize returns the estimated size of the sstable being written if
 	// a call to Close() was made without adding additional keys.
 	EstimatedSize() uint64
-	// ComparePrev compares the provided user to the last point key written to the
-	// writer. The returned value is equivalent to Compare(key, prevKey) where
-	// prevKey is the last point key written to the writer.
+	// ComparePrev compares the provided user key to the last point key written
+	// to the writer. The returned value is equivalent to Compare(key, prevKey)
+	// where prevKey is the last point key written to the writer.
 	//
 	// If no key has been written yet, ComparePrev returns +1.
 	//
@@ -354,6 +354,13 @@ type RawWriter interface {
 	// Metadata returns the metadata for the finished sstable. Only valid to
 	// call after the sstable has been finished.
 	Metadata() (*WriterMetadata, error)
+	// IsPrefixEqualPrev compares the provided user key's prefix to the key
+	// prefix of the last point key written to the writer.
+	//
+	// If no key has been written yet, IsPrefixEqualPrev returns false.
+	//
+	// Must not be called after Writer is closed.
+	IsPrefixEqualPrev(k []byte) bool
 
 	// rewriteSuffixes rewrites the table's data blocks to all contain the
 	// provided suffix. It's specifically used for the implementation of

testdata/compaction/l0_to_lbase_compaction

@@ -29,7 +29,6 @@ L6:
   000006:[itbr@1#159646,SET-rmbh@1#319226,SET]
   000007:[rmbi@1#319227,SET-zzzz@1#475263,SET]
 
-
 metrics
 ----
 ----

testdata/compaction/mvcc_garbage_blob

@@ -0,0 +1,28 @@
+# Set the minimum size for a separated value to 5.
+
+define value-separation=(true, 5, 3, 0s, 1.0)
+----
+
+batch
+set bar bar
+set foo foo
+set fuzz fuzz
+set yaya yaya
+----
+
+batch
+set yay@3 a
+set yay@2 ab
+set zoo@3 b
+set zoo@2 ba
+----
+
+# This flush *should* write a blob file for our MVCC garbage values, containing
+# 2 values: "ab" and "ba" - totaling 4 bytes of logical values.
+
+flush
+----
+L0.0:
+  000005:[bar#10,SET-zoo@2#17,SET] seqnums:[10-17] points:[bar#10,SET-zoo@2#17,SET] size:834 blobrefs:[(B000006: 4); depth:1]
+Blob files:
+  B000006 physical:{000006 size:[94 (94B)] vals:[4 (4B)]}

testdata/compaction/value_separation

@@ -215,7 +215,7 @@ set w world
 ----
 
 # This flush *should* write a blob file, containing 2 values: "hello" and
-# "world" totalling 10 bytes of logical values.
+# "world" totaling 10 bytes of logical values.
 
 flush
 ----

value_separation.go

@@ -172,7 +172,10 @@ func uniqueInputBlobMetadatas(
 }
 
 // writeNewBlobFiles implements the strategy and mechanics for separating values
-// into external blob files.
+// into external blob files. We will always separate potential MVCC garbage
+// values into this external blob file. MVCC garbage values are determined on a
+// best-effort basis; see comments in sstable.IsLikelyMVCCGarbage for the
+// exact criteria we use.
 type writeNewBlobFiles struct {
 	comparer *base.Comparer
 	// newBlobObject constructs a new blob object for use in the compaction.
@@ -197,7 +200,7 @@ type writeNewBlobFiles struct {
 	// short attribute extractor returns an error.
 	invalidValueCallback func(userKey []byte, value []byte, err error)
 
-	// Current blob writer state
+	// Current blob writer state.
 	writer  *blob.FileWriter
 	objMeta objstorage.ObjectMetadata
 
@@ -233,7 +236,7 @@ func (vs *writeNewBlobFiles) EstimatedReferenceSize() uint64 {
 // Add adds the provided key-value pair to the sstable, possibly separating the
 // value into a blob file.
 func (vs *writeNewBlobFiles) Add(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool,
+	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, isLikelyMVCCGarbage func() bool,
 ) error {
 	// We always fetch the value if we're rewriting blob files. We want to
 	// replace any references to existing blob files with references to new blob
@@ -246,10 +249,14 @@ func (vs *writeNewBlobFiles) Add(
 		vs.buf = v[:0]
 	}
 
-	// Values that are too small are never separated.
-	if len(v) < vs.minimumSize {
+	// Values that are too small are never separated; however, MVCC keys are
+	// separated if they are a SET key kind, as long as the value is not empty.
+	//
+	// TODO(annie): Also allow SetWithDelete keys to be separated.
+	if len(v) < vs.minimumSize && !isLikelyMVCCGarbage() {
 		return tw.Add(kv.K, v, forceObsolete)
 	}
+
 	// Merge and deletesized keys are never separated.
 	switch kv.K.Kind() {
 	case base.InternalKeyKindMerge, base.InternalKeyKindDeleteSized:
@@ -405,7 +412,7 @@ func (vs *preserveBlobReferences) EstimatedReferenceSize() uint64 {
 // Add implements compact.ValueSeparation. This implementation will write
 // existing blob references to the output table.
 func (vs *preserveBlobReferences) Add(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool,
+	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ func() bool,
 ) error {
 	if !kv.V.IsBlobValueHandle() {
 		// If the value is not already a blob handle (either it's in-place or in