db: add fallback behavior if ShortAttributeExtractor errors

When separating a value, we call a user-provided 'ShortAttributeExtractor' to
extract a few bits of data from the value. These bits, known as the 'short
attribute', are stored in-place with the sstable key for fast retrieval.

Previously, if the user-provided ShortAttributeExtractor errored, the calling
flush or compaction would abort and bubble the error up. If this was during a
flush, the engine was unable to make progress, repeatedly attempting to flush a
key that cannot be flushed.

This commit updates the logic to fallback to storing a value in-place if the
short attribute extractor is unable to parse a value. The error is still
surfaced through the PossibleAPIMisuse event.

Author: jbowens

compaction.go

@@ -32,6 +32,7 @@ import (
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/cockroachdb/redact"
 )
 
 var errEmptyTable = errors.New("pebble: empty table")
@@ -3258,9 +3259,10 @@ func (d *DB) compactAndWrite(
 		},
 		MissizedDeleteCallback: func(userKey []byte, elidedSize, expectedSize uint64) {
 			d.opts.EventListener.PossibleAPIMisuse(PossibleAPIMisuseInfo{
-				Kind:      MissizedDelete,
-				UserKey:   slices.Clone(userKey),
-				ExtraInfo: fmt.Sprintf("elidedSize=%d,expectedSize=%d", elidedSize, expectedSize),
+				Kind:    MissizedDelete,
+				UserKey: slices.Clone(userKey),
+				ExtraInfo: redact.Sprintf("elidedSize=%d,expectedSize=%d",
+					redact.SafeUint(elidedSize), redact.SafeUint(expectedSize)),
 			})
 		},
 	}

event.go

@@ -646,12 +646,14 @@ type PossibleAPIMisuseInfo struct {
 	// UserKey is set for the following kinds:
 	//  - IneffectualSingleDelete,
 	//  - NondeterministicSingleDelete,
-	//  - MissizedDelete.
+	//  - MissizedDelete,
+	//  - InvalidValue.
 	UserKey []byte
 
 	// ExtraInfo is set for the following kinds:
 	//  - MissizedDelete: contains "elidedSize=<size>,expectedSize=<size>"
-	ExtraInfo string
+	//  - InvalidValue: contains "callback=<callbackName>,value=<value>,err=<err>"
+	ExtraInfo redact.RedactableString
 }
 
 func (i PossibleAPIMisuseInfo) String() string {
@@ -664,7 +666,9 @@ func (i PossibleAPIMisuseInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 	case IneffectualSingleDelete, NondeterministicSingleDelete:
 		w.Printf("possible API misuse: %s (key=%q)", redact.Safe(i.Kind), i.UserKey)
 	case MissizedDelete:
-		w.Printf("possible API misuse: %s (key=%q, %s)", redact.Safe(i.Kind), i.UserKey, redact.Safe(i.ExtraInfo))
+		w.Printf("possible API misuse: %s (key=%q, %s)", redact.Safe(i.Kind), i.UserKey, i.ExtraInfo)
+	case InvalidValue:
+		w.Printf("possible API misuse: %s (key=%q, %s)", redact.Safe(i.Kind), i.UserKey, i.ExtraInfo)
 	default:
 		if invariants.Enabled {
 			panic("invalid API misuse event")
@@ -759,6 +763,12 @@ const (
 	// not accurately record the size of the value it deleted. This can lead to
 	// incorrect behavior in compactions.
 	MissizedDelete
+
+	// InvalidValue is emitted when a user-implemented callback (such as
+	// ShortAttributeExtractor) returns an error for a committed value. This
+	// suggests that either the callback is not implemented for all possible
+	// values or a malformed value was committed to the DB.
+	InvalidValue
 )
 
 func (k APIMisuseKind) String() string {
@@ -769,6 +779,8 @@ func (k APIMisuseKind) String() string {
 		return "nondeterministic SINGLEDEL"
 	case MissizedDelete:
 		return "missized DELSIZED"
+	case InvalidValue:
+		return "invalid value"
 	default:
 		return "unknown"
 	}

testdata/value_separation_policy

@@ -155,3 +155,27 @@ no blob file created
 blobrefs:[
  0: B000003 5
 ]
+
+# Test handling of a short attribute extractor that errors when we attempt to
+# extract the short attribute. We should fall back to writing the value inplace
+# within the output sstable.
+
+init write-new-blob-files minimum-size=2 short-attr-extractor=error
+----
+
+add
+bar#201,SET:poi
+bax#202,SET:blob{value=yaya}
+----
+# create: 000007.sst
+# invalid value for key "bar", value: "poi": short attribute extractor error
+RawWriter.Add("bar#201,SET", "poi", false)
+# invalid value for key "bax", value: "yaya": short attribute extractor error
+RawWriter.Add("bax#202,SET", "yaya", false)
+
+close-output
+----
+# sync-data: 000007.sst
+# close: 000007.sst
+no blob file created
+blobrefs:[]

value_separation.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
+	"github.com/cockroachdb/redact"
 )
 
 var neverSeparateValues getValueSeparation = func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation {
@@ -56,6 +57,16 @@ func (d *DB) determineCompactionValueSeparation(
 		shortAttrExtractor: d.opts.Experimental.ShortAttributeExtractor,
 		writerOpts:         d.opts.MakeBlobWriterOptions(c.outputLevel.level),
 		minimumSize:        policy.MinimumSize,
+		invalidValueCallback: func(userKey []byte, value []byte, err error) {
+			// The value may not be safe, so it will be redacted when redaction
+			// is enabled.
+			d.opts.EventListener.PossibleAPIMisuse(PossibleAPIMisuseInfo{
+				Kind:    InvalidValue,
+				UserKey: userKey,
+				ExtraInfo: redact.Sprintf("callback=ShortAttributeExtractor,value=%x,err=%q",
+					value, err),
+			})
+		},
 	}
 }
 
@@ -167,6 +178,9 @@ type writeNewBlobFiles struct {
 	// to the sstable (but may still be written to a value block within the
 	// sstable).
 	minimumSize int
+	// invalidValueCallback is called when a value is encountered for which the
+	// short attribute extractor returns an error.
+	invalidValueCallback func(userKey []byte, value []byte, err error)
 
 	// Current blob writer state
 	writer  *blob.FileWriter
@@ -230,7 +244,15 @@ func (vs *writeNewBlobFiles) Add(
 		keyPrefixLen := vs.comparer.Split(kv.K.UserKey)
 		shortAttr, err = vs.shortAttrExtractor(kv.K.UserKey, keyPrefixLen, v)
 		if err != nil {
-			return err
+			// Report that there was a value for which the short attribute
+			// extractor was unable to extract a short attribute.
+			vs.invalidValueCallback(kv.K.UserKey, v, err)
+
+			// Rather than erroring out and aborting the flush or compaction, we
+			// fallback to writing the value verbatim to the sstable. Otherwise
+			// a flush could busy loop, repeatedly attempting to write the same
+			// memtable and repeatedly unable to extract a key's short attribute.
+			return tw.Add(kv.K, v, forceObsolete)
 		}
 	}
 

value_separation_test.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/blobtest"
 	"github.com/cockroachdb/pebble/internal/compact"
@@ -94,8 +95,19 @@ func TestValueSeparationPolicy(t *testing.T) {
 							fn++
 							return objStore.Create(ctx, base.FileTypeBlob, fn, objstorage.CreateOptions{})
 						},
+						invalidValueCallback: func(userKey []byte, value []byte, err error) {
+							fmt.Fprintf(&buf, "# invalid value for key %q, value: %q: %s\n", userKey, value, err)
+						},
 					}
 					d.MaybeScanArgs(t, "minimum-size", &newSep.minimumSize)
+					if arg, ok := d.Arg("short-attr-extractor"); ok {
+						switch arg.SingleVal(t) {
+						case "error":
+							newSep.shortAttrExtractor = errShortAttrExtractor
+						default:
+							t.Fatalf("unknown short attribute extractor: %s", arg.String())
+						}
+					}
 					vs = newSep
 				default:
 					t.Fatalf("unknown value separation policy: %s", x)
@@ -153,6 +165,13 @@ func TestValueSeparationPolicy(t *testing.T) {
 		})
 }
 
+func errShortAttrExtractor(key []byte, keyPrefixLen int, value []byte) (ShortAttribute, error) {
+	return 0, errors.New("short attribute extractor error")
+}
+
+// Assert that errShortAttrExtractor implements the ShortAttributeExtractor
+var _ ShortAttributeExtractor = errShortAttrExtractor
+
 // loggingRawWriter wraps a sstable.RawWriter and logs calls to Add and
 // AddWithBlobHandle to provide observability into the separation of values into
 // blob files.