pebble, sstable: preserve blob values only if inputs match output policy

Now that we have persisted the value separation policy used by each sstable,
we can now decide to only preserve blob references during a compaction if the
compaction's value separation policy matches that of its inputs.

Resolves: #4814

Author: xinhaoz

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) compact.ValueSeparation
+	getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) compact.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) compact.ValueSeparation
+type getValueSeparation func(JobID, *tableCompaction, ValueStoragePolicy) compact.ValueSeparation
 
 // newCompaction constructs a compaction from the provided picked compaction.
 //
@@ -3300,10 +3300,7 @@ func (d *DB) runDefaultTableCompaction(
 	d.mu.Unlock()
 	defer d.mu.Lock()
 
-	// Determine whether we should separate values into blob files.
-	valueSeparation := c.getValueSeparation(jobID, c)
-
-	result := d.compactAndWrite(jobID, c, snapshots, valueSeparation)
+	result := d.compactAndWrite(jobID, c, snapshots)
 	if result.Err == nil {
 		ve, result.Err = c.makeVersionEdit(result)
 	}
@@ -3344,10 +3341,7 @@ func (d *DB) runDefaultTableCompaction(
 // compactAndWrite runs the data part of a compaction, where we set up a
 // compaction iterator and use it to write output tables.
 func (d *DB) compactAndWrite(
-	jobID JobID,
-	c *tableCompaction,
-	snapshots compact.Snapshots,
-	valueSeparation compact.ValueSeparation,
+	jobID JobID, c *tableCompaction, snapshots compact.Snapshots,
 ) (result compact.Result) {
 	suggestedCacheReaders := blob.SuggestedCachedReaders(len(c.inputs))
 	// Compactions use a pool of buffers to read blocks, avoiding polluting the
@@ -3442,25 +3436,36 @@ func (d *DB) compactAndWrite(
 	}
 	runner := compact.NewRunner(runnerCfg, iter)
 
-	var spanPolicyValid bool
-	var spanPolicy SpanPolicy
-	// If spanPolicyValid is true and spanPolicyEndKey is empty, then spanPolicy
-	// applies for the rest of the keyspace.
+	// 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 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
+	}
 
+	valueSeparation := c.getValueSeparation(jobID, c, compactionValueStoragePolicy)
 	for runner.MoreDataToWrite() {
 		if c.cancel.Load() {
 			return runner.Finish().WithError(ErrCancelledCompaction)
 		}
 		// Create a new table.
 		firstKey := runner.FirstKey()
-		if !spanPolicyValid || (len(spanPolicyEndKey) > 0 && d.cmp(firstKey, spanPolicyEndKey) >= 0) {
+		if 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)

data_test.go

@@ -912,7 +912,10 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 	// all the tables, we can construct all the referenced blob files and add
 	// them to the final version edit.
 	valueSeparator := &defineDBValueSeparator{
-		pbr:   &preserveBlobReferences{},
+		pbr: &preserveBlobReferences{
+			originalValueSeprationKind: sstable.ValueSeparationDefault,
+			minimumValueSize:           d.opts.Experimental.ValueSeparationPolicy().MinimumSize,
+		},
 		metas: make(map[base.BlobFileID]*manifest.PhysicalBlobFile),
 	}
 
@@ -936,7 +939,7 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 		if err != nil {
 			return err
 		}
-		c.getValueSeparation = func(JobID, *tableCompaction) compact.ValueSeparation {
+		c.getValueSeparation = func(JobID, *tableCompaction, ValueStoragePolicy) compact.ValueSeparation {
 			return valueSeparator
 		}
 		// NB: define allows the test to exactly specify which keys go

internal/manifest/table_metadata.go

@@ -413,6 +413,13 @@ type TableBackingProperties struct {
 	TombstoneDenseBlocksRatio float64
 
 	CompressionStats block.CompressionStats
+
+	// ValueSeparationKind is the value separation policy used when writing the table.
+	ValueSeparationKind sstable.ValueSeparationKind
+	// ValueSeparationMinSize is the minimum value size for which values were
+	// separated when writing the table. This value is 0 if the policy used
+	// does not write blob files.
+	ValueSeparationMinSize uint64
 }
 
 // NumPointDeletions is the number of point deletions in the sstable. For virtual
@@ -447,6 +454,8 @@ func (b *TableBacking) PopulateProperties(props *sstable.Properties) *TableBacki
 		NumRangeKeyDels:            props.NumRangeKeyDels,
 		NumRangeKeySets:            props.NumRangeKeySets,
 		ValueBlocksSize:            props.ValueBlocksSize,
+		ValueSeparationKind:        sstable.ValueSeparationKind(props.ValueSeparationKind),
+		ValueSeparationMinSize:     props.ValueSeparationMinSize,
 	}
 	if props.NumDataBlocks != 0 {
 		b.props.TombstoneDenseBlocksRatio = float64(props.NumTombstoneDenseBlocks) / float64(props.NumDataBlocks)

internal/manifest/table_metadata_test.go

@@ -198,7 +198,7 @@ func TestTableMetadataSize(t *testing.T) {
 			structSize, tableMetadataSize)
 	}
 
-	const tableBackingSize = 160
+	const tableBackingSize = 176
 	if structSize := unsafe.Sizeof(TableBacking{}); structSize != tableBackingSize {
 		t.Errorf("TableBacking struct size (%d bytes) is not expected size (%d bytes)",
 			structSize, tableBackingSize)

sstable/properties.go

@@ -247,10 +247,6 @@ const (
 	// ValueSeparationSpanPolicy indicates that values were separated
 	// based on a span policy during writing.
 	ValueSeparationSpanPolicy
-
-	// ValueSeparationPreservedRefs indicates that existing blob references
-	// were preserved without creating new blob files.
-	ValueSeparationPreservedRefs
 )
 
 var (