Repository: cockroachdb/pebble (go)

Issue:
db: tie value separation policy to span policy
The Options allow configuring a `SpanPolicyFunc` to dictate a particular policy to be maintained over a span of the keyspace. This includes a `ValueStoragePolicy`:

```
// ValueStoragePolicy is a hint used to determine where to store the values for
// KVs.
type ValueStoragePolicy uint8

const (
	// ValueStorageDefault is the default value; Pebble will respect global
	// configuration for value blocks and value separation.
	ValueStorageDefault ValueStoragePolicy = iota

	// ValueStorageLowReadLatency indicates Pebble should prefer storing values
	// in-place.
	ValueStorageLowReadLatency
)
```

We'd like the ability to configure a value separation policy that applies just to the raft log. The raft log is rarely ever read from the storage engine, so it can tolerate separation of its values. We can reduce write amplification through this separation.

We probably should add a `ValueStorageConserveWriteBandwidth` or the like enum option, and then separately add a new Options config that specifies the `ValueSeparationPolicy` to use for spans that use that enum variant.



Jira issue: PEBBLE-529

Pull request description:
db: support overriding value separation minimum-size by key span
Allow the user to configure that a keys within a particular key-range are tolerant of high latency of value retrieval, indicating that their values should be more aggressively separated when value separation is enabled.

We intend to use this in CockroachDB to separate the values of KVs within the raft log. Typically raft log entries are never read because the relevant KVs are cached in-memory until the log is truncated.

Close #5017.

Patch:
diff --git a/compaction.go b/compaction.go
index bf6a3e41..1c91a75b 100644
--- a/compaction.go
+++ b/compaction.go
@@ -3397,8 +3397,15 @@ func (d *DB) compactAndWrite(
 			writerOpts.Compression = block.FastestCompression
 		}
 		vSep := valueSeparation
-		if spanPolicy.ValueStoragePolicy == ValueStorageLowReadLatency {
+		switch spanPolicy.ValueStoragePolicy {
+		case ValueStorageLowReadLatency:
 			vSep = compact.NeverSeparateValues{}
+		case ValueStorageLatencyTolerant:
+			// This span of keyspace is more tolerant of latency, so set a more
+			// aggressive value separation policy for this output.
+			vSep.SetNextOutputConfig(compact.ValueSeparationOutputConfig{
+				MinimumSize: latencyTolerantMinimumSize,
+			})
 		}
 		objMeta, tw, err := d.newCompactionOutputTable(jobID, c, writerOpts)
 		if err != nil {
diff --git a/internal/compact/run.go b/internal/compact/run.go
index 1c5392d4..6e921668 100644
--- a/internal/compact/run.go
+++ b/internal/compact/run.go
@@ -110,9 +110,21 @@ type RunnerConfig struct {
 	GrantHandle base.CompactionGrantHandle
 }
 
+// ValueSeparationOutputConfig is used to configure value separation for an
+// individual compaction output.
+type ValueSeparationOutputConfig struct {
+	// MinimumSize is the minimum size of a value that will be separated into a
+	// blob file.
+	MinimumSize int
+}
+
 // ValueSeparation defines an interface for writing some values to separate blob
 // files.
 type ValueSeparation interface {
+	// SetNextOutputConfig is called when a compaction is starting a new output
+	// sstable. It can be used to configure value separation specifically for
+	// the next compaction output.
+	SetNextOutputConfig(config ValueSeparationOutputConfig)
 	// EstimatedFileSize returns an estimate of the disk space consumed by the
 	// current, pending blob file if it were closed now. If no blob file has
 	// been created, it returns 0.
@@ -475,6 +487,9 @@ type NeverSeparateValues struct{}
 // Assert that NeverSeparateValues implements the ValueSeparation interface.
 var _ ValueSeparation = NeverSeparateValues{}
 
+// SetNextOutputConfig implements the ValueSeparation interface.
+func (NeverSeparateValues) SetNextOutputConfig(config ValueSeparationOutputConfig) {}
+
 // EstimatedFileSize implements the ValueSeparation interface.
 func (NeverSeparateValues) EstimatedFileSize() uint64 { return 0 }
 
diff --git a/options.go b/options.go
index 30a0fb14..fe4a23bb 100644
--- a/options.go
+++ b/options.go
@@ -10,6 +10,7 @@ import (
 	"io"
 	"regexp"
 	"runtime"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -1210,6 +1211,24 @@ type SpanPolicy struct {
 	ValueStoragePolicy ValueStoragePolicy
 }
 
+// String returns a string representation of the SpanPolicy.
+func (p SpanPolicy) String() string {
+	var sb strings.Builder
+	if p.PreferFastCompression {
+		sb.WriteString("fast-compression,")
+	}
+	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,")
+	}
+	return strings.TrimSuffix(sb.String(), ",")
+}
+
 // ValueStoragePolicy is a hint used to determine where to store the values for
 // KVs.
 type ValueStoragePolicy uint8
@@ -1222,6 +1241,15 @@ const (
 	// ValueStorageLowReadLatency 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
 )
 
 // SpanPolicyFunc is used to determine the SpanPolicy for a key region.
@@ -1237,32 +1265,56 @@ const (
 // keyspace after startKey.
 type SpanPolicyFunc func(startKey []byte) (policy SpanPolicy, endKey []byte, err error)
 
+// SpanAndPolicy defines a key range and the policy to apply to it.
+type SpanAndPolicy struct {
+	KeyRange KeyRange
+	Policy   SpanPolicy
+}
+
 // MakeStaticSpanPolicyFunc returns a SpanPolicyFunc that applies a given policy
-// to the given span (and the default policy outside the span).
-func MakeStaticSpanPolicyFunc(cmp base.Compare, span KeyRange, policy SpanPolicy) SpanPolicyFunc {
+// to the given span (and the default policy outside the span). The supplied
+// policies must be non-overlapping in key range.
+func MakeStaticSpanPolicyFunc(cmp base.Compare, inputPolicies ...SpanAndPolicy) SpanPolicyFunc {
+	// Collect all the boundaries of the input policies, sort and deduplicate them.
+	uniqueKeys := make([][]byte, 0, 2*len(inputPolicies))
+	for i := range inputPolicies {
+		uniqueKeys = append(uniqueKeys, inputPolicies[i].KeyRange.Start)
+		uniqueKeys = append(uniqueKeys, inputPolicies[i].KeyRange.End)
+	}
+	slices.SortFunc(uniqueKeys, cmp)
+	uniqueKeys = slices.CompactFunc(uniqueKeys, func(a, b []byte) bool { return cmp(a, b) == 0 })
+
+	// Create a list of policies.
+	policies := make([]SpanPolicy, len(uniqueKeys)-1)
+	for _, p := range inputPolicies {
+		idx, _ := slices.BinarySearchFunc(uniqueKeys, p.KeyRange.Start, cmp)
+		policies[idx] = p.Policy
+	}
+
 	return func(startKey []byte) (_ SpanPolicy, endKey []byte, _ error) {
-		if cmp(startKey, span.End) >= 0 {
-			//      Start     End
-			//      v         v
-			// -----|---------|-----|---
-			//                      ^
-			//                      startKey
+		// Find the policy that applies to the start key.
+		idx, eq := slices.BinarySearchFunc(uniqueKeys, startKey, cmp)
+		switch idx {
+		case len(uniqueKeys):
+			// The start key is after the last policy.
 			return SpanPolicy{}, nil, nil
+		case len(uniqueKeys) - 1:
+			if eq {
+				// The start key is exactly the start of the last policy.
+				return SpanPolicy{}, nil, nil
+			}
+		case 0:
+			if !eq {
+				// The start key is before the first policy.
+				return SpanPolicy{}, uniqueKeys[0], nil
+			}
 		}
-		if cmp(startKey, span.Start) < 0 {
-			//      Start     End
-			//      v         v
-			// --|--|---------|-----
-			//   ^
-			//   startKey
-			return SpanPolicy{}, span.Start, nil
+		if eq {
+			// The start key is exactly the start of this policy.
+			return policies[idx], uniqueKeys[idx+1], nil
 		}
-		//      Start     End
-		//      v         v
-		// -----|----|----|-----
-		//           ^
-		//           startKey
-		return policy, span.End, nil
+		// The start key is between two policies.
+		return policies[idx-1], uniqueKeys[idx], nil
 	}
 }
 
diff --git a/value_separation.go b/value_separation.go
index 99ef914b..9024d7a1 100644
--- a/value_separation.go
+++ b/value_separation.go
@@ -19,6 +19,11 @@ import (
 	"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, sstable.TableFormat) compact.ValueSeparation {
 	return compact.NeverSeparateValues{}
 }
@@ -58,6 +63,7 @@ func (d *DB) determineCompactionValueSeparation(
 		shortAttrExtractor: d.opts.Experimental.ShortAttributeExtractor,
 		writerOpts:         d.opts.MakeBlobWriterOptions(c.outputLevel.level),
 		minimumSize:        policy.MinimumSize,
+		globalMinimumSize:  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.
@@ -178,7 +184,15 @@ type writeNewBlobFiles struct {
 	// separated into a blob file. Values smaller than this are always written
 	// to the sstable (but may still be written to a value block within the
 	// sstable).
+	//
+	// minimumSize is set to globalMinimumSize by default and on every call to
+	// FinishOutput. It may be overriden by SetNextOutputConfig (i.e, if a
+	// SpanPolicy dictates a different minimum size for a span of the keyspace).
 	minimumSize int
+	// globalMinimumSize is the size threshold for separating values into blob
+	// files globally across the keyspace. It may be overridden per-output by
+	// SetNextOutputConfig.
+	globalMinimumSize 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)
@@ -193,6 +207,11 @@ type writeNewBlobFiles struct {
 // Assert that *writeNewBlobFiles implements the compact.ValueSeparation interface.
 var _ compact.ValueSeparation = (*writeNewBlobFiles)(nil)
 
+// SetNextOutputConfig implements the ValueSeparation interface.
+func (vs *writeNewBlobFiles) SetNextOutputConfig(config compact.ValueSeparationOutputConfig) {
+	vs.minimumSize = config.MinimumSize
+}
+
 // EstimatedFileSize returns an estimate of the disk space consumed by the current
 // blob file if it were closed now.
 func (vs *writeNewBlobFiles) EstimatedFileSize() uint64 {
@@ -311,6 +330,8 @@ func (vs *writeNewBlobFiles) FinishOutput() (compact.ValueSeparationMetadata, er
 		ValueSize:    stats.UncompressedValueBytes,
 		CreationTime: uint64(time.Now().Unix()),
 	}
+	// Reset the minimum size for the next output.
+	vs.minimumSize = vs.globalMinimumSize
 
 	return compact.ValueSeparationMetadata{
 		BlobReferences: manifest.BlobReferences{manifest.MakeBlobReference(
@@ -359,6 +380,9 @@ type pendingReference struct {
 // interface.
 var _ compact.ValueSeparation = (*preserveBlobReferences)(nil)
 
+// SetNextOutputConfig implements the ValueSeparation interface.
+func (vs *preserveBlobReferences) SetNextOutputConfig(config compact.ValueSeparationOutputConfig) {}
+
 // EstimatedFileSize returns an estimate of the disk space consumed by the current
 // blob file if it were closed now.
 func (vs *preserveBlobReferences) EstimatedFileSize() uint64 {


Test patch:
diff --git a/data_test.go b/data_test.go
index 0c764414..10d80a43 100644
--- a/data_test.go
+++ b/data_test.go
@@ -1644,6 +1644,7 @@ func describeLSM(d *DB, verbose bool) string {
 }
 
 func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
+	var spanPolicies []SpanAndPolicy
 	for _, cmdArg := range args {
 		switch cmdArg.Key {
 		case "auto-compactions":
@@ -1762,7 +1763,23 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 				DisableValueSeparationBySuffix: true,
 				ValueStoragePolicy:             ValueStorageLowReadLatency,
 			}
-			opts.Experimental.SpanPolicyFunc = MakeStaticSpanPolicyFunc(opts.Comparer.Compare, span, policy)
+			spanPolicies = append(spanPolicies, SpanAndPolicy{
+				KeyRange: span,
+				Policy:   policy,
+			})
+		case "latency-tolerant-span":
+			if len(cmdArg.Vals) != 2 {
+				return errors.New("latency-tolerant-span expects 2 arguments: <start-key> <end-key>")
+			}
+			span := KeyRange{
+				Start: []byte(cmdArg.Vals[0]),
+				End:   []byte(cmdArg.Vals[1]),
+			}
+			policy := SpanPolicy{ValueStoragePolicy: ValueStorageLatencyTolerant}
+			spanPolicies = append(spanPolicies, SpanAndPolicy{
+				KeyRange: span,
+				Policy:   policy,
+			})
 		case "target-file-sizes":
 			if len(cmdArg.Vals) > len(opts.Levels) {
 				return errors.New("too many target-file-sizes")
@@ -1820,6 +1837,9 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 			opts.WALFailover.EnsureDefaults()
 		}
 	}
+	if len(spanPolicies) > 0 {
+		opts.Experimental.SpanPolicyFunc = MakeStaticSpanPolicyFunc(opts.Comparer.Compare, spanPolicies...)
+	}
 	return nil
 }
 
diff --git a/options_test.go b/options_test.go
index f38f3c52..45a54944 100644
--- a/options_test.go
+++ b/options_test.go
@@ -5,14 +5,18 @@
 package pebble
 
 import (
+	"bytes"
+	"fmt"
 	"math/rand/v2"
 	"runtime"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/strparse"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
@@ -602,3 +606,40 @@ func TestApplyDBCompressionSettings(t *testing.T) {
 	require.Equal(t, block.FastestCompression, o.Levels[2].Compression())
 	require.Equal(t, block.FastestCompression, o.Levels[3].Compression())
 }
+
+func TestStaticSpanPolicyFunc(t *testing.T) {
+	var buf bytes.Buffer
+	datadriven.RunTest(t, "testdata/static_span_policy_func", func(t *testing.T, td *datadriven.TestData) string {
+		buf.Reset()
+		var inputSpanPolicies []SpanAndPolicy
+		for _, cmdArg := range td.CmdArgs {
+			p := strparse.MakeParser("-:", cmdArg.String())
+			var sap SpanAndPolicy
+			sap.KeyRange.Start = []byte(p.Next())
+			p.Expect("-")
+			sap.KeyRange.End = []byte(p.Next())
+			p.Expect(":")
+			switch tok := p.Next(); tok {
+			case "lowlatency":
+				sap.Policy.ValueStoragePolicy = ValueStorageLowReadLatency
+			case "latencytolerant":
+				sap.Policy.ValueStoragePolicy = ValueStorageLatencyTolerant
+			default:
+				t.Fatalf("unknown policy: %s", tok)
+			}
+			inputSpanPolicies = append(inputSpanPolicies, sap)
+		}
+
+		spf := MakeStaticSpanPolicyFunc(testkeys.Comparer.Compare, inputSpanPolicies...)
+		for _, key := range strings.Fields(td.Input) {
+			policy, endKey, err := spf([]byte(key))
+			require.NoError(t, err)
+			if endKey == nil {
+				fmt.Fprintf(&buf, "%s -> none\n", key)
+			} else {
+				fmt.Fprintf(&buf, "%s -> %s until %s\n", key, policy, endKey)
+			}
+		}
+		return buf.String()
+	})
+}
diff --git a/testdata/compaction/value_separation b/testdata/compaction/value_separation
index 52f5cea7..0296f828 100644
--- a/testdata/compaction/value_separation
+++ b/testdata/compaction/value_separation
@@ -535,3 +535,116 @@ would excise 1 files.
   add-table:     L6 000053(000044):[a@1#0,SET-azzz@1#0,SET] seqnums:[0-0] points:[a@1#0,SET-azzz@1#0,SET] size:242560(715291) blobrefs:[(B000006: 556894), (B000008: 556959), (B000010: 68494); depth:1]
   add-table:     L6 000054(000044):[c@1#0,SET-czms@1#0,SET] seqnums:[0-0] points:[c@1#0,SET-czms@1#0,SET] size:230320(715291) blobrefs:[(B000006: 528792), (B000008: 528854), (B000010: 65037); depth:1]
   add-backing:   000044
+
+
+# Test a value separation policy that is configured to only separate values ≥
+# 1024 bytes, but there's also a key span defined with the latency-tolerant
+# value storage policy.
+
+define value-separation=(true,1024,10,0s,1.0) latency-tolerant-span=(f,o)
+----
+
+batch
+set a hello
+set b helloworld
+set c helloworld!
+set d hello
+set e helloworld
+set f helloworld!
+set g hello
+set h helloworld
+set i helloworld!
+set j hello
+set k helloworld
+set l helloworld!
+set m hello
+set n helloworld
+set o helloworld!
+set p hello
+set q helloworld
+set r helloworld!
+set s hello
+set t helloworld
+set u helloworld!
+set v hello
+set w helloworld
+set x helloworld!
+set y hello
+set z helloworld
+----
+
+flush
+----
+L0.0:
+  000005:[a#10,SET-e#14,SET] seqnums:[10-14] points:[a#10,SET-e#14,SET] size:799
+  000006:[f#15,SET-n#23,SET] seqnums:[15-23] points:[f#15,SET-n#23,SET] size:925 blobrefs:[(B000007: 63); depth:1]
+  000008:[o#24,SET-z#35,SET] seqnums:[24-35] points:[o#24,SET-z#35,SET] size:836
+Blob files:
+  B000007 physical:{000007 size:[115 (115B)] vals:[63 (63B)]}
+
+iter
+first
+next
+next
+stats
+next
+next
+next
+stats
+next
+next
+next
+stats
+next
+next
+next
+stats
+next
+next
+next
+stats
+next
+next
+next
+stats
+next
+next
+next
+stats
+next
+next
+next
+stats
+----
+a: (hello, .)
+b: (helloworld, .)
+c: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 2 times (2 internal); blocks: 158B cached; points: 3 (3B keys, 26B values)
+d: (hello, .)
+e: (helloworld, .)
+f: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 5 times (5 internal); blocks: 421B cached; points: 6 (6B keys, 43B values); separated: 1 (11B, 11B fetched)
+g: (hello, .)
+h: (helloworld, .)
+i: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 8 times (8 internal); blocks: 421B cached; points: 9 (9B keys, 52B values); separated: 3 (32B, 32B fetched)
+j: (hello, .)
+k: (helloworld, .)
+l: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 11 times (11 internal); blocks: 421B cached; points: 12 (12B keys, 61B values); separated: 5 (53B, 53B fetched)
+m: (hello, .)
+n: (helloworld, .)
+o: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 14 times (14 internal); blocks: 615B cached; points: 15 (15B keys, 79B values); separated: 6 (63B, 63B fetched)
+p: (hello, .)
+q: (helloworld, .)
+r: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 17 times (17 internal); blocks: 615B cached; points: 18 (18B keys, 105B values); separated: 6 (63B, 63B fetched)
+s: (hello, .)
+t: (helloworld, .)
+u: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 20 times (20 internal); blocks: 615B cached; points: 21 (21B keys, 131B values); separated: 6 (63B, 63B fetched)
+v: (hello, .)
+w: (helloworld, .)
+x: (helloworld!, .)
+stats: seeked 1 times (1 internal); stepped 23 times (23 internal); blocks: 615B cached; points: 24 (24B keys, 157B values); separated: 6 (63B, 63B fetched)
diff --git a/testdata/static_span_policy_func b/testdata/static_span_policy_func
new file mode 100644
index 00000000..4ff4bf36
--- /dev/null
+++ b/testdata/static_span_policy_func
@@ -0,0 +1,59 @@
+# A single policy.
+
+test d-f:lowlatency
+a b c d e f g
+----
+a ->  until d
+b ->  until d
+c ->  until d
+d -> low-read-latency until f
+e -> low-read-latency until f
+f -> none
+g -> none
+
+
+# A single encompassing policy.
+
+test a-z:lowlatency
+a b c d e z
+----
+a -> low-read-latency until z
+b -> low-read-latency until z
+c -> low-read-latency until z
+d -> low-read-latency until z
+e -> low-read-latency until z
+z -> none
+
+# Abutting policies.
+
+test b-d:lowlatency d-f:latencytolerant f-h:lowlatency
+a b c d e f g h i z
+----
+a ->  until b
+b -> low-read-latency until d
+c -> low-read-latency until d
+d -> latency-tolerant until f
+e -> latency-tolerant until f
+f -> low-read-latency until h
+g -> low-read-latency until h
+h -> none
+i -> none
+z -> none
+
+# Gaps between policies.
+
+test b-d:lowlatency h-j:latencytolerant
+a b c d e f g h i j k l
+----
+a ->  until b
+b -> low-read-latency until d
+c -> low-read-latency until d
+d ->  until h
+e ->  until h
+f ->  until h
+g ->  until h
+h -> latency-tolerant until j
+i -> latency-tolerant until j
+j -> none
+k -> none
+l -> none
diff --git a/value_separation_test.go b/value_separation_test.go
index aad132a4..e5987804 100644
--- a/value_separation_test.go
+++ b/value_separation_test.go
@@ -100,6 +100,7 @@ func TestValueSeparationPolicy(t *testing.T) {
 						},
 					}
 					d.MaybeScanArgs(t, "minimum-size", &newSep.minimumSize)
+					newSep.globalMinimumSize = newSep.minimumSize
 					if arg, ok := d.Arg("short-attr-extractor"); ok {
 						switch arg.SingleVal(t) {
 						case "error":
@@ -206,6 +207,9 @@ type defineDBValueSeparator struct {
 // Assert that *defineDBValueSeparator implements the compact.ValueSeparation interface.
 var _ compact.ValueSeparation = (*defineDBValueSeparator)(nil)
 
+// SetNextOutputConfig implements the compact.ValueSeparation interface.
+func (vs *defineDBValueSeparator) SetNextOutputConfig(config compact.ValueSeparationOutputConfig) {}
+
 // EstimatedFileSize returns an estimate of the disk space consumed by the current
 // blob file if it were closed now.
 func (vs *defineDBValueSeparator) EstimatedFileSize() uint64 {