db, valsep: make span ValueStoragePolicy a struct defining the min size

ValueStoragePolicy in SpanPolicy is now a struct defining the policy adjustment
type (defaultPolicy, noValSeparation, override) and the override value for the
minimum size used for value separation. Compactions now use the
ValueStoragePolicy returned by the span policy for a table's key range to decide
whether we can preserve blob references or not.

Previously, we would rewrite blob references if a compaction contained tables
with mixed value storage policies, even if those policies were unchanged.

We now preserve blob references if:
- The maximum blob reference depth is not exceeded.
- For each input table, the value separation policy used when writing the table
matches the current value separation policy applied to the table span.

Author: xinhaoz

blob_rewrite_test.go

@@ -92,8 +92,7 @@ func TestBlobRewrite(t *testing.T) {
 					pbr := valsep.NewPreserveAllHotBlobReferences(
 						inputBlobPhysicalFiles,
 						0, /* outputBlobReferenceDepth */
-						sstable.ValueSeparationDefault,
-						0, /* original minimumSize */
+						0, /* minimumSize */
 					)
 					vs = pbr
 				case "write-new-blob-files":

compaction.go

@@ -250,7 +250,7 @@ type tableCompaction struct {
 	// b) rewrite blob files: The compaction will write eligible values to new
 	// blob files. This consumes more write bandwidth because all values are
 	// rewritten. However it restores locality.
-	getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation
+	getValueSeparation func(JobID, *tableCompaction) valsep.ValueSeparation
 
 	// startLevel is the level that is being compacted. Inputs from startLevel
 	// and outputLevel will be merged to produce a set of outputLevel files.
@@ -542,7 +542,7 @@ func (c *tableCompaction) makeInfo(jobID JobID) CompactionInfo {
 	return info
 }
 
-type getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation
+type getValueSeparation func(JobID, *tableCompaction) valsep.ValueSeparation
 
 // newCompaction constructs a compaction from the provided picked compaction.
 //
@@ -3416,38 +3416,27 @@ func (d *DB) compactAndWrite(
 	}
 	runner := compact.NewRunner(runnerCfg, iter)
 
-	// Determine the value separation policy for this compaction.
-	// We pass the value storage span policy if one applies for the entire
-	// compaction keyspace in order to preserve blob references.
-	// Note that the value storage policy may change per table if
-	// there are different span policies in effect for this output
-	// range.
-	var compactionValueStoragePolicy ValueStoragePolicy
-	var spanPolicyEndKey []byte
+	var spanPolicyValid bool
 	var spanPolicy SpanPolicy
-	spanPolicy, spanPolicyEndKey, err = d.opts.Experimental.SpanPolicyFunc(c.bounds.Start)
-	if err != nil {
-		return runner.Finish().WithError(err)
-	}
-	if len(spanPolicyEndKey) == 0 || d.cmp(c.bounds.End.Key, spanPolicyEndKey) < 0 {
-		compactionValueStoragePolicy = spanPolicy.ValueStoragePolicy
-	}
+	// If spanPolicyValid is true and spanPolicyEndKey is empty, then spanPolicy
+	// applies for the rest of the keyspace.
+	var spanPolicyEndKey []byte
 
-	valueSeparation := c.getValueSeparation(jobID, c, compactionValueStoragePolicy)
+	valueSeparation := c.getValueSeparation(jobID, c)
 	for runner.MoreDataToWrite() {
 		if c.cancel.Load() {
 			return runner.Finish().WithError(ErrCancelledCompaction)
 		}
 		// Create a new table.
 		firstKey := runner.FirstKey()
-		if len(spanPolicyEndKey) > 0 && d.cmp(firstKey, spanPolicyEndKey) >= 0 {
+		if !spanPolicyValid || (len(spanPolicyEndKey) > 0 && d.cmp(firstKey, spanPolicyEndKey) >= 0) {
 			var err error
 			spanPolicy, spanPolicyEndKey, err = d.opts.Experimental.SpanPolicyFunc(firstKey)
 			if err != nil {
 				return runner.Finish().WithError(err)
 			}
+			spanPolicyValid = true
 		}
-
 		writerOpts := d.makeWriterOptions(c.outputLevel.level)
 		if spanPolicy.DisableValueSeparationBySuffix {
 			writerOpts.DisableValueBlocks = true
@@ -3456,14 +3445,16 @@ func (d *DB) compactAndWrite(
 			writerOpts.Compression = block.FastestCompression
 		}
 		vSep := valueSeparation
-		switch spanPolicy.ValueStoragePolicy {
-		case ValueStorageLowReadLatency:
+		switch spanPolicy.ValueStoragePolicy.PolicyAdjustment {
+		case UseDefaultValueStorage:
+			// No change to value separation.
+		case NoValueSeparation:
 			vSep = valsep.NeverSeparateValues{}
-		case ValueStorageLatencyTolerant:
+		case OverrideValueStorage:
 			// This span of keyspace is more tolerant of latency, so set a more
 			// aggressive value separation policy for this output.
 			vSep.SetNextOutputConfig(valsep.ValueSeparationOutputConfig{
-				MinimumSize: latencyTolerantMinimumSize,
+				MinimumSize: spanPolicy.ValueStoragePolicy.MinimumSize,
 			})
 		}
 		objMeta, tw, err := d.newCompactionOutputTable(jobID, c, writerOpts)

compaction_test.go

@@ -1507,6 +1507,48 @@ func runCompactionTest(
 			s := blobRewriteLog.String()
 			return s
 
+		case "set-span-policies":
+			var spanPolicies []SpanAndPolicy
+			for _, line := range strings.Split(td.Input, "\n") {
+				line = strings.TrimSpace(line)
+				args := strings.Fields(line)
+				if len(args) < 2 {
+					td.Fatalf(t, "expected span and policy, got: %s", line)
+				}
+				// First arg should be keys in the form <start>,<end>
+				keys := strings.Split(args[0], ",")
+				keyRange := KeyRange{
+					Start: []byte(keys[0]),
+					End:   []byte(keys[1]),
+				}
+				policy := SpanPolicy{}
+				args = args[1:]
+				for _, arg := range args {
+					parts := strings.Split(arg, "=")
+					switch parts[0] {
+					case "val-sep-minimum-size":
+						if len(parts) != 2 {
+							td.Fatalf(t, "expected val-sep-minimum-size=<size>, got: %s", arg)
+						}
+						size, err := strconv.ParseUint(parts[1], 10, 64)
+						if err != nil {
+							td.Fatalf(t, "parsing value-minimum-size: %s", err)
+						}
+						policy.ValueStoragePolicy.MinimumSize = int(size)
+						if size == 0 {
+							policy.ValueStoragePolicy.PolicyAdjustment = NoValueSeparation
+						} else if int(size) != d.opts.Experimental.ValueSeparationPolicy().MinimumSize {
+							policy.ValueStoragePolicy.PolicyAdjustment = OverrideValueStorage
+						}
+					}
+				}
+				spanPolicies = append(spanPolicies, SpanAndPolicy{
+					KeyRange: keyRange,
+					Policy:   policy,
+				})
+			}
+			d.opts.Experimental.SpanPolicyFunc = MakeStaticSpanPolicyFunc(d.cmp, spanPolicies...)
+			return ""
 		default:
 			return fmt.Sprintf("unknown command: %s", td.Cmd)
 		}

compaction_value_separation.go

@@ -9,17 +9,11 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/objstorage"
-	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/valsep"
 	"github.com/cockroachdb/redact"
 )
 
-// latencyTolerantMinimumSize is the minimum size, in bytes, of a value that
-// will be separated into a blob file when the value storage policy is
-// ValueStorageLatencyTolerant.
-const latencyTolerantMinimumSize = 10
-
-var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation {
+var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction) valsep.ValueSeparation {
 	return valsep.NeverSeparateValues{}
 }
 
@@ -29,7 +23,7 @@ var neverSeparateValues getValueSeparation = func(JobID, *tableCompaction, Value
 //
 // It assumes that the compaction will write tables at d.TableFormat() or above.
 func (d *DB) determineCompactionValueSeparation(
-	jobID JobID, c *tableCompaction, valueStorage ValueStoragePolicy,
+	jobID JobID, c *tableCompaction,
 ) valsep.ValueSeparation {
 	if d.FormatMajorVersion() < FormatValueSeparation ||
 		d.opts.Experimental.ValueSeparationPolicy == nil {
@@ -48,25 +42,12 @@ func (d *DB) determineCompactionValueSeparation(
 		// For flushes, c.version is nil.
 		blobFileSet = uniqueInputBlobMetadatas(&c.version.BlobFiles, c.inputs)
 	}
-	minSize := policy.MinimumSize
-	switch valueStorage {
-	case ValueStorageLowReadLatency:
-		return valsep.NeverSeparateValues{}
-	case ValueStorageLatencyTolerant:
-		minSize = latencyTolerantMinimumSize
-	default:
-	}
-	if writeBlobs, outputBlobReferenceDepth := shouldWriteBlobFiles(c, policy, uint64(minSize)); !writeBlobs {
+	if writeBlobs, outputBlobReferenceDepth := shouldWriteBlobFiles(c, policy, d.opts.Experimental.SpanPolicyFunc, d.cmp); !writeBlobs {
 		// This compaction should preserve existing blob references.
-		kind := sstable.ValueSeparationDefault
-		if valueStorage != ValueStorageDefault {
-			kind = sstable.ValueSeparationSpanPolicy
-		}
 		return valsep.NewPreserveAllHotBlobReferences(
 			blobFileSet,
 			outputBlobReferenceDepth,
-			kind,
-			minSize,
+			policy.MinimumSize,
 		)
 	}
 
@@ -101,7 +82,7 @@ func (d *DB) determineCompactionValueSeparation(
 // maximum blob reference depth to assign to output sstables (the actual value
 // may be lower iff the output table references fewer distinct blob files).
 func shouldWriteBlobFiles(
-	c *tableCompaction, policy ValueSeparationPolicy, minimumValueSizeForCompaction uint64,
+	c *tableCompaction, policy ValueSeparationPolicy, spanPolicyFunc SpanPolicyFunc, cmp Compare,
 ) (writeBlobs bool, referenceDepth manifest.BlobReferenceDepth) {
 	// Flushes will have no existing references to blob files and should write
 	// their values to new blob files.
@@ -124,6 +105,7 @@ func shouldWriteBlobFiles(
 		// We should try to write to new blob files.
 		return true, 0
 	}
+
 	// If the compaction's output blob reference depth would be greater than the
 	// configured max, we should rewrite the values into new blob files to
 	// restore locality.
@@ -140,7 +122,32 @@ func shouldWriteBlobFiles(
 			if !backingPropsValid {
 				continue
 			}
-			if backingProps.ValueSeparationMinSize != minimumValueSizeForCompaction {
+
+			var expectedMinSize int
+			bounds := t.UserKeyBounds()
+			spanPolicy, spanPolicyEndKey, err := spanPolicyFunc(bounds.Start)
+			if err != nil {
+				// For now, if we can't determine the span policy, we should just assume
+				// the default policy is in effect for this table.
+				expectedMinSize = policy.MinimumSize
+			} else {
+				if len(spanPolicyEndKey) > 0 && cmp(bounds.End.Key, spanPolicyEndKey) >= 0 {
+					// The table's key range now uses multiple span policies. Rewrite to new
+					// blob files so values are stored according to the current policy.
+					return true, 0
+				}
+				switch spanPolicy.ValueStoragePolicy.PolicyAdjustment {
+				case UseDefaultValueStorage:
+					// Use the global policy's minimum size.
+					expectedMinSize = policy.MinimumSize
+				case OverrideValueStorage:
+					expectedMinSize = spanPolicy.ValueStoragePolicy.MinimumSize
+				case NoValueSeparation:
+					expectedMinSize = 0
+				}
+			}
+
+			if int(backingProps.ValueSeparationMinSize) != expectedMinSize {
 				return true, 0
 			}
 		}

data_test.go

@@ -918,7 +918,6 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 	valueSeparator.pbr = valsep.NewPreserveAllHotBlobReferences(
 		valueSeparator.metas,
 		0, /* outputreference depth */
-		sstable.ValueSeparationDefault,
 		d.opts.Experimental.ValueSeparationPolicy().MinimumSize,
 	)
 
@@ -937,7 +936,7 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 			flushable: mem,
 			flushed:   make(chan struct{}),
 		}}
-		getValueSeparator := func(JobID, *tableCompaction, ValueStoragePolicy) valsep.ValueSeparation {
+		getValueSeparator := func(JobID, *tableCompaction) valsep.ValueSeparation {
 			return valueSeparator
 		}
 		c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.latest.l0Organizer,

options.go

@@ -1294,7 +1294,7 @@ type SpanPolicy struct {
 
 	// ValueStoragePolicy is a hint used to determine where to store the values
 	// for KVs.
-	ValueStoragePolicy ValueStoragePolicy
+	ValueStoragePolicy ValueStoragePolicyAdjustment
 }
 
 // String returns a string representation of the SpanPolicy.
@@ -1306,38 +1306,64 @@ func (p SpanPolicy) String() string {
 	if p.DisableValueSeparationBySuffix {
 		sb.WriteString("disable-value-separation-by-suffix,")
 	}
-	switch p.ValueStoragePolicy {
-	case ValueStorageLowReadLatency:
-		sb.WriteString("low-read-latency,")
-	case ValueStorageLatencyTolerant:
-		sb.WriteString("latency-tolerant,")
+	switch p.ValueStoragePolicy.PolicyAdjustment {
+	case NoValueSeparation:
+		sb.WriteString("no-value-separation,")
+	case OverrideValueStorage:
+		sb.WriteString("override,")
 	}
 	return strings.TrimSuffix(sb.String(), ",")
 }
 
-// ValueStoragePolicy is a hint used to determine where to store the values for
-// KVs.
-type ValueStoragePolicy uint8
+// ValueStoragePolicyAdjustment is used to determine where to store the values for
+// KVs. If the PolicyAdjustment specified is OverrideValueStorage, the remaining fields
+// are used to override the global configuration for value separation.
+type ValueStoragePolicyAdjustment struct {
+	// PolicyAdjustment specifies the policy adjustment to apply.
+	PolicyAdjustment ValueStoragePolicyAdjustmentType
+
+	// Remaining fields are ignored, unless the PolicyAdjustment is OverrideValueStorage.
+
+	// MinimumSize is the minimum size of the value.
+	MinimumSize int
+}
+
+// ValueStoragePolicyAdjustmentType is a hint used to determine where to store the
+// values for KVs.
+type ValueStoragePolicyAdjustmentType uint8
 
 const (
-	// ValueStorageDefault is the default value; Pebble will respect global
-	// configuration for value blocks and value separation.
-	ValueStorageDefault ValueStoragePolicy = iota
+	// UseDefaultValueStorage is the default value; Pebble will respect global
+	// configuration for value separation.
+	UseDefaultValueStorage ValueStoragePolicyAdjustmentType = iota
 
-	// ValueStorageLowReadLatency indicates Pebble should prefer storing values
+	// NoValueSeparation indicates Pebble should prefer storing values
 	// in-place.
-	ValueStorageLowReadLatency
-
-	// ValueStorageLatencyTolerant indicates value retrieval can tolerate
-	// additional latency, so Pebble should aggressively prefer storing values
-	// separately if it can reduce write amplification.
-	//
-	// If the global Options' enable value separation, Pebble may choose to
-	// separate values under the LatencyTolerant policy even if they do not meet
-	// the minimum size threshold of the global Options' ValueSeparationPolicy.
-	ValueStorageLatencyTolerant
+	NoValueSeparation
+
+	// OverrideValueStorage indicates that value separation thresholds (see
+	// valsep.ValueSeparationOutputConfig) for this key range are being
+	// overridden from a SpanPolicy. If the global Options enable value
+	// separation, Pebble will separate values under the OverrideValueStorage
+	// policy even if they do not meet the minimum size threshold of the
+	// global Options' ValueSeparationPolicy.
+	OverrideValueStorage
 )
 
+// ValueStorageLatencyTolerant is the suggested ValueStoragePolicyAdjustment
+// to use for key ranges that can tolerate higher value retrieval
+// latency.
+var ValueStorageLatencyTolerant = ValueStoragePolicyAdjustment{
+	PolicyAdjustment: OverrideValueStorage,
+	MinimumSize:      10,
+}
+
+// ValueStorageLowReadLatency is the suggested ValueStoragePolicyAdjustment
+// to use for key ranges that require low value retrieval latency.
+var ValueStorageLowReadLatency = ValueStoragePolicyAdjustment{
+	PolicyAdjustment: NoValueSeparation,
+}
+
 // SpanPolicyFunc is used to determine the SpanPolicy for a key region.
 //
 // The returned policy is valid from the start key until (and not including) the

sstable/properties.go

@@ -241,7 +241,8 @@ const (
 	ValueSeparationNone ValueSeparationKind = iota
 
 	// ValueSeparationDefault indicates that values were separated into
-	// new blob files during writing.
+	// new blob files during writing using the default value separation
+	// policy settings for minimum sizes.
 	ValueSeparationDefault
 
 	// ValueSeparationSpanPolicy indicates that values were separated